Spark：保存到 HDFS 时出现内存不足错误

Question

我在将大数据保存到 hdfs 时遇到 OOME

val accumulableCollection = sc.accumulableCollection(ArrayBuffer[String]())
val rdd = textfile.filter(row => {
    if (row.endsWith(",")) {
        accumulableCollection += row
        false
    } else if (row.length < 100) {
        accumulableCollection += row
        false
    }
    valid
})
rdd.cache()
val rdd2 = rdd.map(_.split(","))
val rdd3 = rdd2.filter(row => {
    var valid = true
    for((k,v) <- fieldsMap if valid ) {
        if (StringUtils.isBlank(row(k)) || "NULL".equalsIgnoreCase(row(k))) {
            accumulableCollection += row.mkString(",")
            valid = false
        }
    }
    valid
})
sc.parallelize(accumulableCollection.value).saveAsTextFile(hdfsPath)

我在 spark-submit 中使用这个：

--num-executors 2 --driver-memory 1G --executor-memory 1G --executor-cores 2

这是日志的输出：

15/04/12 18:46:49 WARN scheduler.TaskSetManager: Stage 4 contains a task of very large size (37528 KB). The maximum recommended task size is 100 KB.
15/04/12 18:46:49 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 4.0 (TID 8, worker4, PROCESS_LOCAL, 38429279 bytes)
15/04/12 18:46:49 INFO scheduler.TaskSetManager: Starting task 1.0 in stage 4.0 (TID 9, worker3, PROCESS_LOCAL, 38456846 bytes)
15/04/12 18:46:50 INFO scheduler.TaskSetManager: Starting task 2.0 in stage 4.0 (TID 10, worker4, PROCESS_LOCAL, 38426488 bytes)
15/04/12 18:46:51 INFO scheduler.TaskSetManager: Starting task 3.0 in stage 4.0 (TID 11, worker3, PROCESS_LOCAL, 38445061 bytes)
15/04/12 18:46:51 INFO cluster.YarnClusterScheduler: Cancelling stage 4
15/04/12 18:46:51 INFO cluster.YarnClusterScheduler: Stage 4 was cancelled
15/04/12 18:46:51 INFO scheduler.DAGScheduler: Job 4 failed: saveAsTextFile at WriteToHdfs.scala:87, took 5.713617 s
15/04/12 18:46:51 INFO yarn.ApplicationMaster: Final app status: FAILED, exitCode: 15, (reason: User class threw exception: Job aborted due to stage failure: Serialized task 8:0 was 38617206 bytes, which exceeds max allowed: spark.akka.frameSize (10485760 bytes) - reserved (204800 bytes). Consider increasing spark.akka.frameSize or using broadcast variables for large values.)
Exception in thread "Driver" org.apache.spark.SparkException: Job aborted due to stage failure: **Serialized task 8:0 was 30617206 bytes, which exceeds max allowed: spark.akka.frameSize (10485760 bytes)** - reserved (204800 bytes). Consider increasing spark.akka.frameSize or using broadcast variables for large values.

序列化任务 8:0 为 30617206 字节，超过允许的最大值：spark.akka.frameSize (10485760 字节) --- (1) 什么是 30MB 序列化任务？

考虑对大值使用广播变量。 --- (2) 广播变量应该是什么？ rdd2?还是 accumulableCollection，因为这是我要写入 HDFS 的内容？

当我增加 frameSize 时，现在的错误是：java.lang.OutOfMemoryError: Java heap space，所以我必须将驱动程序内存和执行程序内存增加到 2G 才能工作。如果 accumulableCollection.value.length 是 500,000 我需要使用 3G。这正常吗？

该文件只有 146MB，包含 200,000 行（对于 2G 内存）。 （在 HDFS 中它分为 2 个分区，每个分区包含 73MB）

Answer 1

Spark 中的核心编程抽象是一个 RDD，you can create them in two ways：

(1) 并行化驱动程序中的现有集合，或 (2) 引用外部存储系统中的数据集，例如共享 文件系统、HDFS、HBase 或任何提供 Hadoop 的数据源 输入格式。

parallelize() 方法 (1) 要求您将整个数据集保存在一台机器的内存中（第 26 页学习火花）。

方法（2），简称External Datasets，适用于大文件。

以下行使用accumulableCollection.value 的内容创建一个RDD，并要求它适合单台机器：

sc.parallelize(accumulableCollection.value)

缓存 RDD 时也可能会超出内存：

rdd.cache()

这意味着整个textfile RDD 都存储在内存中。您很可能不想这样做。请参阅Spark documentation，了解为数据选择缓存级别的建议。

Answer 2

它的意思几乎就是它所说的。您正在尝试序列化一个非常大的对象。您可能应该重写您的代码以不这样做。

例如，我不清楚您为什么要尝试更新可累积集合，并在 filter 中这样做，它甚至可以执行多次。然后您缓存 RDD，但您已经尝试在驱动程序上有一个副本？然后你将其他值添加到本地集合中，然后再次将其转换为 RDD？

为什么是 accumulableCollection？只需对 RDD 进行操作。这里有很多冗余。