我认为使用过滤器和Map从原始数据创建派生数据流就足够了：

val numericFeaturesDStream = originalDStream.filter(e => predicate(e)).map(e => extractFeatures(e))
val otherNumericFeaturesDStream = originalDStream.filter(e => predicate2(e)).map(e => extractOtherFeatures(e))

注意这些 filter 以及 map 这些步骤可以结合在一起 collect 步骤（不要与无参数rdd.collect混淆，它将数据传送到驱动程序！！！）

val featuresStream = originalDStream.transform(rdd => 
  rdd.collect{case (a,b,c,d,e) if c=="client" => Vectors.parse(a)}
)
streamingKMeans.trainOn(featuresStream)

我们还可以将一组动态过滤的数据流保存到某个集合中。这里我们使用一个包含我们用来过滤的键的Map：

originalDStream.cache() // important for performance when a DStream is branched out.
// filterKeys: Set[String] 
val dstreamByFilterKey = filterKeys.map(key => key -> originalDStream.filter(e => (getKey(e)==key)))
// do something with the different DStreams in this structure ...

这些代码片段是用实际逻辑完成的代码示例。