scala - Scala Spark 中的 NullPointerException 似乎是由集合类型引起的？

Question

sessionIdList 的类型为:

scala> sessionIdList
res19: org.apache.spark.rdd.RDD[String] = MappedRDD[17] at distinct at <console>:30

当我尝试运行以下代码时:

val x = sc.parallelize(List(1,2,3)) 
val cartesianComp = x.cartesian(x).map(x => (x))

val kDistanceNeighbourhood = sessionIdList.map(s => {
    cartesianComp.filter(v => v != null)
})

kDistanceNeighbourhood.take(1)

我收到异常:

14/05/21 16:20:46 ERROR Executor: Exception in task ID 80
java.lang.NullPointerException
        at org.apache.spark.rdd.RDD.filter(RDD.scala:261)
        at $line94.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$1.apply(<console>:38)
        at $line94.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$1.apply(<console>:36)
        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)

但是如果我使用:

val l = sc.parallelize(List("1","2")) 
val kDistanceNeighbourhood = l.map(s => {    
    cartesianComp.filter(v => v != null)
})

kDistanceNeighbourhood.take(1)

然后没有显示异常

两个代码片段之间的区别在于第一个片段中的 sessionIdList 类型为:

res19: org.apache.spark.rdd.RDD[String] = MappedRDD[17] at distinct at <console>:30

在第二个片段中“l”是类型

scala> l
res13: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[32] at parallelize at <console>:12

为什么会出现这个错误？

我是否需要将 sessionIdList 转换为 ParallelCollectionRDD 才能解决此问题？

Answer 1

Spark 不支持 RDD 嵌套(请参阅 https://stackoverflow.com/a/14130534/590203 了解同一问题的另一情况)，因此您无法在其他 RDD 操作内部对 RDD 执行转换或操作。

在第一种情况下，当工作线程尝试访问仅存在于驱动程序而非工作线程上的 SparkContext 对象时，您会看到工作线程抛出 NullPointerException。

在第二种情况下，我的预感是该作业是在驱动程序上本地运行的，并且纯粹是偶然的。