Apache Spark фильтр некоторые

Question

У меня есть следующая leftOuterJoin операция:

val totalsAndProds = transByProd.leftOuterJoin(products)
println(totalsAndProds.first())

, которая печатает:

(19,([Ljava.lang.String;@261ea657,Some([Ljava.lang.String;@25290bca)))

, затем я пытаюсь применить следующие filter операции:

totalsAndProds.filter(x => x._2 == Some).first

но происходит сбой со следующим исключением:

Exception in thread "main" java.lang.UnsupportedOperationException: empty collection
    at org.apache.spark.rdd.RDD$$anonfun$first$1.apply(RDD.scala:1380)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
    at org.apache.spark.rdd.RDD.first(RDD.scala:1377)
    at com.example.spark.WordCount$.main(WordCount.scala:98)
    at com.example.spark.WordCount.main(WordCount.scala)

что я делаю не так, и операция фильтра возвращает пустую коллекцию?

Answer 1

Ваш предикат неверен:

1. Ваш тип RDD (Int, (Array[String], Option[Array[String]])), поэтому _._2 имеет тип (Array[String], Option[Array[String]]), а не Option[Array[String]]
2. Вы не проверяетеТипы опций с использованием равенства.

Попробуйте

totalsAndProds.filter{ case (_, (_, s)) => s.isDefined }

Пример ниже:

scala> val rdd = sc.parallelize(List((19, (Array("a"), Some(Array("a"))))))
rdd: org.apache.spark.rdd.RDD[(Int, (Array[String], Some[Array[String]]))] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> rdd.filter{ case (_, (_, s)) => s.isDefined }
res0: org.apache.spark.rdd.RDD[(Int, (Array[String], Some[Array[String]]))] = MapPartitionsRDD[1] at filter at <console>:27

scala> rdd.filter{ case (_, (_, s)) => s.isDefined }.collect
res1: Array[(Int, (Array[String], Some[Array[String]]))] = Array((19,(Array(a),Some([Ljava.lang.String;@5307fee))))