使用saveAsTextFile的Spark NullPointerException

Question

在尝试合并并保存RDD时,我得到了一个N​​PE.

代码在本地工作,并在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 动作被称为上RDD.
因此,当您执行转换时,它只是作为要执行的操作存储.

该saveAsTextFile方法是一个动作,同时地图操作是转换.

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

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