如何在apache spark中将RDD [ParentClass]与RDD [Subclass]匹配？

Question

我必须将 rdd 与其类型相匹配。

trait Fruit

case class Apple(price:Int) extends Fruit
case class Mango(price:Int) extends  Fruit

现在DStream[Fruit] 类型的 dstream 即将到来。它是Apple 或Mango。

如何根据子类进行操作？类似下面的东西（不起作用）：

dStream.foreachRDD{rdd:RDD[Fruit] =>
     rdd match {
       case rdd: RDD[Apple] =>
         //do something

       case rdd: RDD[Mango] =>
         //do something

       case _ =>
         println(rdd.count() + "<<<< not matched anything")
     }
    }

Answer 1

由于我们有RDD[Fruit]，因此任何行都可以是Apple 或Mango。使用foreachRDD 时，每个RDD 将包含这些（以及可能的其他）类型的混合。

为了区分不同的类型，我们可以使用collect[U](f: PartialFunction[T, U]): RDD[U]（不要与collect(): Array[T] 混淆，collect(): Array[T] 返回包含来自 RDD 的元素的列表）。 该函数将通过应用函数f返回一个包含所有匹配值的RDD（在这种情况下，我们可以在这里使用模式匹配）。

下面是一个小的说明性示例（也将Orange 添加到水果中）。

设置：

val ssc = new StreamingContext(spark.sparkContext, Seconds(1))
val inputData: Queue[RDD[Fruit]] = Queue()
val dStream: InputDStream[Fruit] = ssc.queueStream(inputData)

inputData += spark.sparkContext.parallelize(Seq(Apple(5), Apple(5), Mango(11)))
inputData += spark.sparkContext.parallelize(Seq(Mango(10), Orange(1), Orange(3)))

这会创建一个 RDD[Fruit] 流，其中包含两个单独的 RDDs。

dStream.foreachRDD{rdd: RDD[Fruit] =>
  val mix = rdd.collect{
    case row: Apple => ("APPLE", row.price) // do any computation on apple rows
    case row: Mango => ("MANGO", row.price) // do any computation on mango rows
    //case _@row => do something with other rows (will be removed by default).
  }
  mix foreach println
}

在上面的collect 中，我们稍微更改每一行（删除类），然后打印结果RDD。结果：

// First RDD
(MANGO,11)
(APPLE,5)
(APPLE,5)

// Second RDD
(MANGO,10)

可以看出，模式匹配保留并更改了包含Apple 和Mango 的行，同时删除了所有Orange 类。

---

单独的 RDD

如果需要，也可以将两个子类分成自己的RDDs，如下所示。然后可以对这两个RDDs 执行任何计算。

val apple = rdd.collect{case row: Apple => row}
val mango = rdd.collect{case row: Mango => row}

---

完整示例代码

trait Fruit
case class Apple(price:Int) extends Fruit
case class Mango(price:Int) extends  Fruit
case class Orange(price:Int) extends  Fruit

object Test {
  def main(args: Array[String]) {
    val spark = SparkSession.builder.master("local[*]").getOrCreate()

    val ssc = new StreamingContext(spark.sparkContext, Seconds(1))
    val inputData: Queue[RDD[Fruit]] = Queue()
    val inputStream: InputDStream[Fruit] = ssc.queueStream(inputData)

    inputData += spark.sparkContext.parallelize(Seq(Apple(5), Apple(5), Mango(11)))
    inputData += spark.sparkContext.parallelize(Seq(Mango(10), Orange(1), Orange(3)))

    inputStream.foreachRDD{rdd:RDD[Fruit] =>
      val mix = rdd.collect{
        case row: Apple => ("APPLE", row.price) // do any computation on apple rows
        case row: Mango => ("MANGO", row.price) // do any computation on mango rows
        //case _@row => do something with other rows (will be removed by default).
      }
      mix foreach println
    }

    ssc.start()
    ssc.awaitTermination()
  }
}