使用 saveAsTextFile 引发 NullPointerException

Question

我在尝试合并和保存 RDD 时遇到 NPE。

代码在本地工作，并且在 scala shell 中的集群上工作，但在将其作为作业提交到集群时抛出错误。

我尝试使用 take() 打印出来以查看 rdd 是否包含一些空数据，但这会引发相同的错误 - 痛苦，因为它在 shell 中工作正常。

我正在保存到 HDFS 并在变量中保存完整的 url 路径 - 模型在 MLLib 训练阶段使用这种方法保存得很好。

任何想法都非常感谢！

Scala 代码（整体预测函数）：

//Load the Random Forest
val rfModel = RandomForestModel.load(sc, modelPath)

//Make the predictions - Here the label is the unique ID of the point
val rfPreds = labDistVect.map(p => (p.label, rfModel.predict(p.features)))


//Collect and save
println("Done Modelling, now saving preds")
val outP = rfPreds.coalesce(1,true).saveAsTextFile(outPreds)
println("Done Modelling, now saving coords")
val outC = coords.coalesce(1,true).saveAsTextFile(outCoords)

堆栈跟踪：

    Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 6.0 failed 4 times, most recent failure: Lost task 0.3 in stage 6.0 (TID 40, XX.XX.XX.XX): java.lang.NullPointerException
    at GeoDistPredict1$$anonfun$38.apply(GeoDist1.scala:340)
    at GeoDistPredict1$$anonfun$38.apply(GeoDist1.scala:340)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:312)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)

Answer 1

Spark 操作分为惰性转换和动作。

当在 RDD 上调用 action 时，会对 RDD 执行 惰性转换。
因此，当您执行转换时，它只是存储为要执行的操作。

saveAsTextFile方法是一个action，map操作是transformation。

如果转换步骤有任何问题，它将在调用转换的操作级别步骤中显示为问题。

因此，您在映射操作期间可能会遇到问题，其中某些字段中存在空值，这很可能导致您的 NPE 问题。