python之执行许多数据帧连接时 PySpark OutOfMemoryErrors_编程设计

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

我在尝试将许多不同的数据帧连接在一起时遇到了 PySpark 中的 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 个较小的数据帧执行连接，这些数据帧有两列 uid 和 data_in_the_form_of_lists(通常是 Python 字符串列表)。我要加入的主数据框有大约 10 列，但还包含一个 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 和 metric3 是我在连接之前将其转换为数据帧的 RDD(请记住，其中实际上有 50 个较小的 metric 我加入的 dfs)。

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

错误是不确定的。我没有看到它们始终出现在我的联接中的任何特定位置，有时似乎出现在我最后的 metrics_df.collect() 调用中，有时出现在较小的联接期间。

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

我还注意到垃圾收集时间有很多:

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

编辑回答评论问题

我一直在将 Spark 作业作为一个更大的 PyCharm 项目的一部分来运行(这就是为什么 spark 上下文围绕着一个类)。我重构了代码以将其作为脚本运行，使用以下 spark 提交:

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

请您参考如下方法:

我遇到了类似的问题并且它适用于:
Spark 提交:

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

代码:

sc = SparkContext().getOrCreate()