Recibo un NPE cuando intento fusionarme y guardar un RDD.

El código funciona localmentey funciona en el clúster en el shell scala, pero arroja el error al enviarlo como un trabajo al clúster.

Intenté imprimir usando una toma () para ver si el rdd contiene algunos datos nulos, pero esto arroja el mismo error: dolor porque funciona bien en el shell.

Estoy guardando en HDFS y tengo la ruta de URL completa en la variable: el modelo guarda bien con este método durante la fase de entrenamiento de MLLib.

Cualquier idea muy apreciada!

Código Scala (Func. De predicción completa):

//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)

Seguimiento de pila:

    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)