读取多个文件时 Spark 驱动程序内存不足答案

【问题标题】：Spark driver running out of memory when reading multiple files读取多个文件时 Spark 驱动程序内存不足
【发布时间】：2018-12-17 06:42:02
【问题描述】：

我的程序是这样工作的：

将大量文件作为数据帧读入。在这些文件中有一组大约 60 个文件，每个文件有 5k 行，我为每个文件创建一个单独的数据框，进行一些简单的处理，然后将它们全部合并到一个数据框中，用于进一步加入。
我最终对多个数据帧执行了多次连接和列计算，最终生成了一个目标数据帧。
我将目标数据框保存为 Parquet 文件。
在同一个 spark 应用程序中，我加载了 Parquet 文件并进行了一些繁重的聚合，然后对该数据帧进行了多次自联接。
我将第二个数据帧保存为另一个 Parquet 文件。

问题

如果我在上面提到的一组文件中只有一个文件而不是 60 个文件，那么一切都适用于具有 8g 内存的驱动程序。对于 60 个文件，前 3 个步骤工作正常，但驱动程序在准备第二个文件时内存不足。只有当我将驱动程序的内存增加到 20g 时，情况才会有所改善。

问题

这是为什么呢？在计算第二个文件时，我不使用用于计算第一个文件的 Dataframe，因此如果第一个 Parquet 文件的大小保持不变，它们的数量和内容应该无关紧要，不是吗？这 60 个数据帧是否会以某种方式被缓存并占用驱动程序的内存？我自己不做任何缓存。我也从不收集任何东西。我不明白为什么 8g 的内存对于 Spark 驱动程序来说是不够的。

【问题讨论】：

嗨，我遇到了同样的问题。我不使用持久化或缓存，看起来镶木地板文件保留在驱动程序内存中。在不增加内存的情况下找到解决方案了吗？
很高兴知道我并不孤单。不，恐怕我还没有找到任何合适的解决方案，除了增加驱动程序内存和读取更少的文件。幸运的是，我能够在我的情况下限制它们。但这并不是一个真正好的长期解决方案（文件的数量无论如何都会随着时间的推移而增长），我仍然不太明白这是由 Spark 中的一些错误还是它对内存的特殊处理引起的，而我没有调整它正确...

标签： apache-spark dataframe apache-spark-sql out-of-memory

【解决方案1】：

缓存或持久性是用于（迭代和交互式）Spark 计算的优化技术。它们有助于保存临时部分结果，以便在后续阶段重复使用。因此，这些作为 RDD 的中间结果保存在内存（默认）或更可靠的存储（如磁盘和/或复制）中。可以使用缓存操作缓存 RDD。它们也可以使用持久化操作来持久化。缓存和持久化操作之间的区别纯粹是语法上的。 cache 是persist 或persist(MEMORY_ONLY) 的同义词，即cache 只是以默认存储级别MEMORY_ONLY 持久化。

参考use of persist and unpersist

【讨论】：

对不起，这对我的问题不是很有用或相关。
你有很多文件要处理，经过几个步骤后内存不足。您需要使用内存管理。使用坚持你的问题将得到解决。 @汤姆
你的理由是什么？我不确定为什么坚持会解决我的问题，因为它本身会使用大量内存。事实上，当持久化数据帧 (MEMORY_AND_DISK_SER) 而不是在第 3 步中保存（然后在最后保存两个数据帧）时，我遇到了 OOM 错误，数据量要少得多。

【解决方案2】：

conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
//you have to use serialization configuration if you are using MEMORY_AND_DISK_SER 


val rdd1 = sc.textFile("some data")
rdd1.persist(storageLevel.MEMORY_AND_DISK_SER)   // marks rdd as persist
val rdd2 = rdd1.filter(...)
val rdd3 = rdd1.map(...)

rdd2.persist(storageLevel.MEMORY_AND_DISK_SER)
rdd3.persist(storageLevel.MEMORY_AND_DISK_SER)

rdd2.saveAsTextFile("...")
rdd3.saveAsTextFile("...")

rdd1.unpersist()
rdd2.unpersist()
rdd3.unpersist()

要调整您的代码，请遵循此link

【讨论】：