不同的调用和地图一起抛出火花库中的NPE

Question

我不确定这是否是一个错误,所以如果你做这样的事情

// d:spark.RDD[String]
d.distinct().map(x => d.filter(_.equals(x)))

你会得到一个Java NPE.但是如果你collect马上做了distinct,一切都会好的.

我正在使用spark 0.6.1.

Answer 1

Spark不支持嵌套RDD或引用其他RDD的用户定义函数,因此NullPointerException; 在spark-users邮件列表上看到这个帖子.

看起来您当前的代码正在尝试d按值对元素进行分组; 您可以使用groupBy() RDD方法有效地执行此操作:

scala> val d = sc.parallelize(Seq("Hello", "World", "Hello"))
d: spark.RDD[java.lang.String] = spark.ParallelCollection@55c0c66a

scala> d.groupBy(x => x).collect()
res6: Array[(java.lang.String, Seq[java.lang.String])] = Array((World,ArrayBuffer(World)), (Hello,ArrayBuffer(Hello, Hello)))