java - Pyspark 中的广播加入得到 OnOutOfMemoryError

Question

我在一个不是很大的数据帧上使用 pyspark 中的广播连接，但出现了 oom 错误。代码非常简单，如下所示:

from pyspark.sql.functions import broadcast
df1 = sc.read.parquet("xxx")
df2 = sc.read.parquet("xxx")
df = df1.join(broadcast(df2), on=["key"])

几秒钟后，我收到了 OOM 错误

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

df2 不是很大，1868895 行。 hdfs文件大约142M。我可以轻松地加载驱动程序代码中的所有数据并将其用作 UDF 没有问题。但是当使用广播连接时，出现OOM。

我的配置如下:

spark.python.worker.memory=2048m
spark.executor.memory=8g
spark.driver.memory=7g

Answer 1

尝试增加 spark.sql.broadcastTimeout 参数的值，
spark 是 SparkSession 对象的地方