Spark Scala代码中的“线程“dispatcher-event-loop-0”java.lang.OutOfMemoryError：Java堆空间中的异常”错误

Question

val data = spark.read
    .text(filePath)
    .toDF("val")
    .withColumn("id", monotonically_increasing_id())



    val count = data.count()



    val header = data.where("id==1").collect().map(s => s.getString(0)).apply(0)



    val columns = header
    .replace("H|*|", "")
    .replace("|##|", "")
    .split("\\|\\*\\|")


    val structSchema = StructType(columns.map(s=>StructField(s, StringType, true)))



    var correctData = data.where('id > 1 && 'id < count-1).select("val")
    var dataString = correctData.collect().map(s => s.getString(0)).mkString("").replace("\\\n","").replace("\\\r","")
    var dataArr = dataString.split("\\|\\#\\#\\|").map(s =>{ 
                                                          var arr = s.split("\\|\\*\\|")
                                                          while(arr.length < columns.length) arr = arr :+ ""
                                                          RowFactory.create(arr:_*)
                                                         })
    val finalDF = spark.createDataFrame(sc.parallelize(dataArr),structSchema)

    display(finalDF)

这部分代码报错：

线程“dispatcher-event-loop-0”java.lang.OutOfMemoryError 中的异常：Java 堆空间

经过几个小时的调试主要是部分：

var dataArr = dataString.split("\\|\\#\\#\\|").map(s =>{ 
                                                          var arr = s.split("\\|\\*\\|")
                                                          while(arr.length < columns.length) arr = arr :+ ""
                                                          RowFactory.create(arr:_*)
                                                         })
    val finalDF = spark.createDataFrame(sc.parallelize(dataArr),structSchema)

导致错误。

我把部分改成

var dataArr = dataString.split("\\|\\#\\#\\|").map(s =>{
                                                          var arr = s.split("\\|\\*\\|")
                                                          while(arr.length < columns.length) arr = arr :+ ""
                                                          RowFactory.create(arr:_*)
                                                         }).toList
  val finalDF = sqlContext.createDataFrame(sc.makeRDD(dataArr),structSchema)

但错误仍然相同。我应该改变什么来避免这种情况？

当我运行此代码是 databricks spark 集群时，特定作业给出了此 Spark 驱动程序错误：

作业因阶段故障而中止：序列化任务 45:0 为 792585456 字节，超过了允许的最大值：spark.rpc.message.maxSize（268435456 字节）。

我添加了这部分代码：

spark.conf.set("spark.rpc.message.maxSize",Int.MaxValue)

但没用。

Answer 1

我的猜测是

var dataString = correctData.collect().map(s => s.getString(0)).mkString("").replace("\\\n","").replace("\\\r","")

是问题所在，因为您将（几乎）所有数据收集到驱动程序，即 1 个单一 JVM。

也许这条线会运行，但对dataString 的后续操作将超出您的内存限制。你不应该收集你的数据！相反，使用分布式“数据结构”，例如 Dataframe 或 RDD。

我认为你可以省略上面一行中的collect