我想使用 spark 将大型（51GB）XML 文件（在外部 HDD 上）读入数据帧（使用spark-xml 插件），进行简单的映射/过滤，重新排序，然后将其写回磁盘，作为 CSV文件。
java.lang.OutOfMemoryError: Java heap space但无论我如何调整它，我总是得到一个。
我想了解为什么增加分区数不会停止OOM错误
它不应该将任务分成更多部分，以便每个单独的部分更小并且不会导致内存问题吗？
（Spark 不可能试图把所有东西都塞进内存中，如果不合适就崩溃了，对吧？？）
我尝试过的事情：
- 在读取和写入时重新分区/合并到（5,000 和 10,000 个分区）数据帧（初始值为 1,604）
- 使用较少数量的执行器（6、4，即使有2 个执行器，我也会收到 OOM 错误！）
- 减小分割文件的大小（默认看起来是 33MB）
- 提供大量内存（我所拥有的）
- 增加到
spark.memory.fraction0.8（默认为 0.6） - 减少
spark.memory.storageFraction到 0.2（默认为 0.5） - 设置
spark.default.parallelism为 30 和 40（我默认为 8） - 设置
spark.files.maxPartitionBytes为 64M（默认为 128M）
我所有的代码都在这里（注意我没有缓存任何东西）：
val df: DataFrame = spark.sqlContext.read
.option("mode", "DROPMALFORMED")
.format("com.databricks.spark.xml")
.schema(customSchema) // defined previously
.option("rowTag", "row")
.load(s"$pathToInputXML")
println(s"\n\nNUM PARTITIONS: ${df.rdd.getNumPartitions}\n\n")
// prints 1604
// i pass `numPartitions` as cli arguments
val df2 = df.coalesce(numPartitions)
// filter and select only the cols i'm interested in
val dsout = df2
.where( df2.col("_TypeId") === "1" )
.select(
df("_Id").as("id"),
df("_Title").as("title"),
df("_Body").as("body"),
).as[Post]
// regexes to clean the text
val tagPat = "<[^>]+>".r
val angularBracketsPat = "><|>|<"
val whitespacePat = """\s+""".r
// more mapping
dsout
.map{
case Post(id,title,body,tags) =>
val body1 = tagPat.replaceAllIn(body,"")
val body2 = whitespacePat.replaceAllIn(body1," ")
Post(id,title.toLowerCase,body2.toLowerCase, tags.split(angularBracketsPat).mkString(","))
}
.orderBy(rand(SEED)) // random sort
.write // write it back to disk
.option("quoteAll", true)
.mode(SaveMode.Overwrite)
.csv(output)
笔记
- 输入拆分非常小（仅 33MB），那么为什么我不能有 8 个线程每个处理一个拆分呢？它真的不应该破坏我的记忆（我已经
更新我编写了一个较短版本的代码，它只读取文件，然后读取 forEachPartition(println)。
我得到同样的OOM错误：
val df: DataFrame = spark.sqlContext.read
.option("mode", "DROPMALFORMED")
.format("com.databricks.spark.xml")
.schema(customSchema)
.option("rowTag", "row")
.load(s"$pathToInputXML")
.repartition(numPartitions)
println(s"\n\nNUM PARTITIONS: ${df.rdd.getNumPartitions}\n\n")
df
.where(df.col("_PostTypeId") === "1")
.select(
df("_Id").as("id"),
df("_Title").as("title"),
df("_Body").as("body"),
df("_Tags").as("tags")
).as[Post]
.map {
case Post(id, title, body, tags) =>
Post(id, title.toLowerCase, body.toLowerCase, tags.toLowerCase))
}
.foreachPartition { rdd =>
if (rdd.nonEmpty) {
println(s"HI! I'm an RDD and I have ${rdd.size} elements!")
}
}
PS：我使用的是 spark v 2.1.0。我的机器有 8 个内核和 16 GB 内存。