scala - 内存不足

Question

我有一个包含 150 G txt 文件的文件夹(大约 700 个文件，平均每个 200 MB)。

我正在使用 Scala 来处理文件并最终计算一些汇总统计数据。我看到两种可能的方法来做到这一点:

手动遍历所有文件，对每个文件进行计算并最终合并结果

将整个文件夹读取到一个 RDD，在这个单个 RDD 上执行所有操作并让 spark 执行所有并行化

我倾向于第二种方法，因为它看起来更简洁(不需要特定于并行化的代码)，但我想知道我的场景是否适合我的硬件和数据强加的约束。我有一台有 16 个线程和 64 GB RAM 可用的工作站(因此并行化将在不同处理器内核之间严格本地化)。稍后我可能会使用更多机器扩展基础设施，但现在我只想专注于调整这个工作站场景的设置。

我正在使用的代码:
- 读取 TSV 文件，并将有意义的数据提取到 (String, String, String) 三元组
- 之后执行一些过滤、映射和分组
- 最后，减少数据并计算一些聚合

我已经能够使用单个文件(约 200 MB 的数据)运行此代码，但是我收到了 java.lang.OutOfMemoryError: GC 开销限制超出
添加更多数据时和/或 Java 堆外异常(应用程序因 6GB 数据中断，但我想将其与 150 GB 数据一起使用)。

我想我必须调整一些参数才能完成这项工作。我将不胜感激有关如何解决此问题的任何提示(如何调试内存需求)。我尝试增加“spark.executor.memory”并使用较少数量的内核(合理的是每个内核都需要一些堆空间)，但这并没有解决我的问题。

我不需要非常快的解决方案(如果需要，它可以轻松运行几个小时甚至几天)。我也没有缓存任何数据，只是最后将它们保存到文件系统中。如果您认为使用手动并行化方法更可行，我也可以这样做。

Answer 1

我和我的团队在 5 台机器 @32GB RAM 上成功处理了大小超过 1 TB 的 csv 数据。这在很大程度上取决于您正在执行哪种处理以及如何处理。

如果您对 RDD 重新分区，则需要额外的计算
有超过你的堆大小的开销，尝试加载更多的文件
通过减少分割大小的并行性TextInputFormat.SPLIT_MINSIZE和 TextInputFormat.SPLIT_MAXSIZE(如果您使用的是 TextInputFormat)来提升
并行性。

尝试使用 mapPartition 而不是 map 以便您可以处理
分区内的计算。如果计算使用临时
变量或实例，您仍然面临内存不足，请尝试
减少每个分区的数据数量(增加分区
号码)

使用增加驱动程序内存和执行程序内存限制
spark中的“spark.executor.memory”和“spark.driver.memory”
创建 Spark Context 前的配置

请注意，Spark 是一个通用的集群计算系统，因此在单台机器上使用 Spark 是低效的(恕我直言)