没有 rdd.contains . 函数 contains 这里使用的是 String 他在房间里 RDD .
比如这里：

val rdd_first = rdd.filter {
  element => element.contains("First") // each `element` is a String 
}

此方法不可靠，因为字符串中的其他内容可能会满足比较要求，从而导致错误。
例如。

{"a":"foo", "b":"bar","type":"second", "c": "first", .......}

处理这个问题的一种方法是首先将json数据转换成适当的记录，然后对这些记录应用分组或过滤逻辑。为此，我们首先需要数据的模式定义。通过该模式，我们可以将记录解析为json，并在此基础上应用任何处理：

case class Record(a:String, b:String, `type`:String)

import org.apache.spark.sql.types._
val schema = StructType(
               Array(
                StructField("a", StringType, true),
                StructField("b", StringType, true),
                StructField("type", String, true)
               )
             )

val processPerType: Map[String, Dataset[Record] => Unit ] = Map(...) 

stream.foreachRDD { rdd =>
  val records = rdd.toDF("value").select(from_json($"value", schema)).as[Record]
  processPerType.foreach{case (tpe, process) =>
      val target = records.filter(entry => entry.`type` == tpe)
      process(target)
  }
}

问题并没有具体说明需要对每种类型的记录应用哪种逻辑。这里介绍的是一种处理任何自定义逻辑都可以表示为函数的问题的通用方法 Dataset[Record] => Unit .
如果逻辑可以表示为聚合，那么 Dataset 聚合函数将更合适。