执行多个数据帧连接时PySpark OutOfMemoryErrors

Question

有很多关于这个问题的帖子，但没有一个回答我的问题。

在尝试将许多不同的数据帧连接在一起时，我在PySpark中遇到了＜code＞OutOfMemoryError

我的本地机器有16GB内存，我已将Spark配置设置为：

class SparkRawConsumer:

    def __init__(self, filename, reference_date, FILM_DATA):
        self.sparkContext = SparkContext(master='local[*]', appName='my_app')
        SparkContext.setSystemProperty('spark.executor.memory', '3g')
        SparkContext.setSystemProperty('spark.driver.memory', '15g')

关于Spark中OOM错误的SO帖子显然很多很多，但基本上大多数都是说增加你的内存属性。

我实际上是对50-60个较小的数据帧执行连接，这些数据帧有两列< code>uid和< code > data _ in _ the _ form _ of _ lists (通常是一个Python字符串列表)。我正在联接的主数据帧大约有10列，但还包含一个< code>uid列(我正在联接)。

我只尝试加入1,500行数据。但是，当显然所有这些数据都可以放入内存时，我会经常遇到OutOfMemory错误。我通过在我的存储中查看我的SparkUI来确认这一点：

在代码中，我的连接如下所示：

# lots of computations to read in my dataframe and produce metric1, metric2, metric3, .... metric 50
metrics_df = metrics_df.join(
                self.sqlContext.createDataFrame(metric1, schema=["uid", "metric1"]), on="uid")

metrics_df.count()
metrics_df.repartition("gid_value")
metrics_df = metrics_df.join(
                self.sqlContext.createDataFrame(metric2, schema=["uid", "metric2"]),
                on="gid_value")

metrics_df.repartition("gid_value")
metrics_df = metrics_df.join(
                self.sqlContext.createDataFrame(metric3, schema=["uid", "metric3"]),
                on="uid")

metrics_df.count()
metrics_df.repartition("gid_value")

其中，metric1、metric2和metic3是我在加入之前转换为数据帧的RDD（请记住，我正在加入的这些较小的metricdfs中实际上有50个）。

我调用metric.count（）来强制评估，因为它似乎有助于防止内存错误（否则在尝试最终收集时，我会得到更多的驱动程序错误）。

这些错误是非确定性的。我没有看到它们始终出现在我的连接中的任何特定位置，有时似乎发生在我最后的metrics_df.collect（）调用中，有时在较小的连接期间。

我真的怀疑任务序列化/反序列化存在一些问题。例如，当我查看典型阶段的事件时间线时，我发现大部分都被任务反序列化占用了：

我还注意到有大量的垃圾回收机制：

垃圾回收是导致内存错误的问题吗？还是任务序列化？

我一直在作为一个更大的PyCharm项目的一部分运行Spark作业（因此Spark上下文被包装在一个类中）。我重构了代码，将其作为脚本运行，使用以下spark提交：

spark-submit spark_consumer.py \
  --driver-memory=10G \
  --executor-memory=5G \
  --conf spark.executor.extraJavaOptions='-XX:+UseParallelGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps'

Answer 1

我面临着类似的问题，它适用于：
Spark Submit：

spark-submit --driver-memory 3g\
            --executor-memory 14g\
            *.py

法典：

sc = SparkContext().getOrCreate()