内存不足错误的 Spark 配置

Question

集群设置 -

Driver has 28gb
Workers have 56gb each (8 workers)

配置-

spark.memory.offHeap.enabled true
spark.driver.memory 20g
spark.memory.offHeap.size 16gb
spark.executor.memory 40g

我的工作 -

//myFunc just takes a string s and does some transformations on it, they are very small strings, but there's about 10million to process.


//Out of memory failure
data.map(s => myFunc(s)).saveAsTextFile(outFile)

//works fine
data.map(s => myFunc(s))

此外，我从我的程序中去集群/删除了 spark，它在具有 56gb 内存的单个服务器上完成得很好（成功保存到文件中）。这表明它只是一个火花配置问题。我查看了https://spark.apache.org/docs/latest/configuration.html#memory-management，我目前的配置似乎是我的工作需要更改的所有内容。我还应该改变什么？

更新-

数据 -

val fis: FileInputStream = new FileInputStream(new File(inputFile))
val bis: BufferedInputStream = new BufferedInputStream(fis);
val input: CompressorInputStream = new CompressorStreamFactory().createCompressorInputStream(bis);
br = new BufferedReader(new InputStreamReader(input))
val stringArray = br.lines().toArray()
val data = sc.parallelize(stringArray)

注意 - 这不会导致任何内存问题，即使它非常低效。我无法使用 spark 读取它，因为它会引发一些 EOF 错误。

myFunc，我不能真正发布它的代码，因为它很复杂。但基本上，输入字符串是一个分隔字符串，它会进行一些分隔符替换、日期/时间规范化等。输出字符串与输入字符串的大小大致相同。

此外，它适用于较小的数据大小，并且输出正确且与输入数据文件的大小大致相同。

Answer 1

您当前的解决方案没有利用火花。您将整个文件加载到内存中的数组中，然后使用 sc.parallelize 将其分发到 RDD 中。这非常浪费内存（即使没有 spark），当然会导致大文件出现内存不足的问题。

改为使用sc.textFile(filePath) 创建您的RDD。然后 spark 能够以块的形式智能地读取和处理文件，因此一次只需要其中一小部分在内存中。您还可以通过这种方式利用并行性，因为 spark 将能够并行读取和处理文件，无论您拥有多少执行程序和代码，而不是需要在单个线程上读取整个文件机器。

假设 myFunc 一次只能查看一行，那么这个程序的内存占用应该非常小。

Answer 2

如果您在 MAP 之前和之后在您的程序中提供更多详细信息，将会有所帮助。 除非触发动作，否则第二个命令（仅 Map）不会执行任何操作。您的文件可能未分区，驱动程序正在执行此工作。下面应该强制将数据均匀地分配给工作人员并保护单个节点上的 OOM。但它会导致数据混洗。

查看您的代码后更新解决方案，如果您这样做会更好

val data = sc.parallelize(stringArray).repartition(8)
data.map(s => myFunc(s)).saveAsTextFile(outFile)