Spark：写入数据帧时“请求的数组大小超过 VM 限制”答案

【问题标题】：Spark: 'Requested array size exceeds VM limit' when writing dataframeSpark：写入数据帧时“请求的数组大小超过 VM 限制”
【发布时间】：2018-10-15 14:27:09
【问题描述】：

我在运行 Scala Spark 作业时遇到“OutOfMemoryError：请求的数组大小超过 VM 限制”错误。

我正在 AWS EMR 集群上运行此作业，其构成如下：

主：1 m4.4xlarge 32 vCore，64 GiB 内存

核心：1 r3.4xlarge 32 vCore，122 GiB 内存

我使用的 Spark 版本是 2.2.1，EMR 发布标签为 5.11.0。

我正在使用以下配置在 spark shell 中运行我的工作：

spark-shell --conf spark.driver.memory=40G 
--conf spark.driver.maxResultSize=25G 
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer 
--conf spark.kryoserializer.buffer.max=2000 
--conf spark.rpc.message.maxSize=2000 
--conf spark.dynamicAllocation.enabled=true

我试图通过这项工作将一列对象数据框转换为包含这些对象列表的单行数据框。

对象如下：

case class Properties (id: String)
case class Geometry (`type`: String, coordinates: Seq[Seq[Seq[String]]])
case class Features (`type`: String, properties: Properties, geometry: Geometry)

我的数据框架构如下：

root
 |-- geometry: struct (nullable = true)
 |    |-- type: string (nullable = true)
 |    |-- coordinates: array (nullable = true)
 |    |    |-- element: array (containsNull = true)
 |    |    |    |-- element: array (containsNull = true)
 |    |    |    |    |-- element: string (containsNull = true)
 |-- type: string (nullable = false)
 |-- properties: struct (nullable = false)
 |    |-- id: string (nullable = true)

我将其转换为列表并将其添加到单行数据框中，如下所示：

val x = Seq(df.collect.toList)
final_df.withColumn("features", typedLit(x))

创建此列表时我没有遇到任何问题，而且速度非常快。但是，当我尝试通过执行以下任一操作将其写出时，此列表的大小似乎存在限制：

final_df.first
final_df.write.json(s"s3a://<PATH>/")

我也尝试通过执行以下操作将列表转换为数据框，但它似乎永远不会结束。

val x = Seq(df.collect.toList)
val y = x.toDF

我能够让这个数据框使用的最大列表有 813318 个 Features 对象，每个对象都包含一个 Geometry 对象，其中包含 33 个元素的列表，总共 29491869 个元素。

在运行我的作业时，尝试编写几乎任何比它更大的列表都会给我以下堆栈跟踪。

# java.lang.OutOfMemoryError: Requested array size exceeds VM limit
# -XX:OnOutOfMemoryError="kill -9 %p"
#   Executing /bin/sh -c "kill -9 33028"...
os::fork_and_exec failed: Cannot allocate memory (12)
18/03/29 21:41:35 ERROR FileFormatWriter: Aborting job null.
java.lang.OutOfMemoryError: Requested array size exceeds VM limit
    at org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder.grow(BufferHolder.java:73)
    at org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter.write(UnsafeArrayWriter.java:217)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply_1$(Unknown Source)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply1_1$(Unknown Source)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
    at org.apache.spark.sql.execution.LocalTableScanExec$$anonfun$unsafeRows$1.apply(LocalTableScanExec.scala:41)
    at org.apache.spark.sql.execution.LocalTableScanExec$$anonfun$unsafeRows$1.apply(LocalTableScanExec.scala:41)
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
    at scala.collection.immutable.List.foreach(List.scala:381)
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
    at scala.collection.immutable.List.map(List.scala:285)
    at org.apache.spark.sql.execution.LocalTableScanExec.unsafeRows$lzycompute(LocalTableScanExec.scala:41)
    at org.apache.spark.sql.execution.LocalTableScanExec.unsafeRows(LocalTableScanExec.scala:36)
    at org.apache.spark.sql.execution.LocalTableScanExec.rdd$lzycompute(LocalTableScanExec.scala:48)
    at org.apache.spark.sql.execution.LocalTableScanExec.rdd(LocalTableScanExec.scala:48)
    at org.apache.spark.sql.execution.LocalTableScanExec.doExecute(LocalTableScanExec.scala:52)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:117)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:117)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:138)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:135)
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:116)
    at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:92)
    at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:92)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply$mcV$sp(FileFormatWriter.scala:173)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:166)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:166)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:65)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:166)
    at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:145)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:58)

我已经尝试进行一百万次配置更改，包括在这项工作中同时投入更多的驱动程序和执行程序内存，但无济于事。有没有办法解决？有什么想法吗？

【问题讨论】：

标签： scala apache-spark spark-dataframe geospatial geojson

【解决方案1】：

问题来了

val x = Seq(df.collect.toList)

当您收集数据帧时，它会将数据帧的所有数据发送给驱动程序。因此，如果您的数据帧很大，这将导致驱动程序内存不足。

需要注意的是，在分配给执行程序的所有内存中，驱动程序可以使用的堆内存通常是 30%（如果不更改）。因此，由于收集操作，驱动程序正在发生数据量阻塞。

现在问题是您可能认为数据帧在磁盘上的大小较小，但那是因为数据已序列化并保存在那里。当您收集它时，它会具体化数据框并使用 JVM 来存储数据。这会导致巨大的内存爆炸（一般为 5-7X）。

我建议您删除收集部分并直接使用df 数据框。因为我侦察

val x = Seq(df.collect.toList) and df are essentially same

【讨论】：

是的，我知道收集是个坏主意，但我需要将原始数据框转换为列表，我不知道有更好的方法来做到这一点。而且我不确定我是否得到最后一部分：Seq(df.collect.toList) 是列表的单个元素列表，但 df 是数据框。你能详细说明它们的相同之处吗？

【解决方案2】：

嗯，有一个数据帧聚合函数可以执行您想要的操作，而无需对驱动程序进行收集。例如，如果您想按键收集所有“特征”列：df.groupBy($"key").agg(collect_list("feature"))，或者如果您真的想对整个数据框执行此操作而不进行分组：df.agg(collect_list("feature"))。

但是，我想知道您为什么要这样做，因为使用每个对象一行的数据框似乎比包含整个结果的单行更容易。即使使用 collect_list 聚合函数，如果您的内存仍然不足，我也不会感到惊讶。

【讨论】：