在 PySpark 中使用 collect_list 时 Java 内存不足

Question

我目前正在使用 PySpark 并在一个包含大约 6 亿条记录的表上运行查询。表本身大约 300GB。我的查询看起来像这样：

select f1, collect_list(struct(f2, f3)) as fdata
from table
group by 1

目前，我收到此错误：

# java.lang.OutOfMemoryError: Java heap space
# -XX:OnOutOfMemoryError="kill -9 %p"
#   Executing /bin/sh -c "kill -9 1010"...
Killed

还有一点是(f2, f3)元组的分布不均匀。一些f1 可能有100k 个这样的元组，而另一些f1 可能只有5 个。我怀疑collect_list 会导致一些问题。

我总是可以增加集群大小，但我想知道我是否可以对表做点什么。按f1 分区也不是一种选择，因为它具有非常高的维度。我对 Spark 或 Hive 没有太多经验，因此我们将不胜感激。

Answer 1

您显然正试图将一个巨大结果集的所有行加载到内存中。

我怀疑 collect_list 引起了一些问题。

这很可能是对的。

解决方案要么是增加堆大小¹，要么是更改处理结果集的方式，以便您不需要全部保存同时在记忆中。

---

^{1 - ... 可能还有更多的内存、更大的机器等。你可以看到这是怎么回事。}