scala - 在 Spark 流中使用过滤器转换后，如何在函数中返回两个 DStream？

Question

在函数中，有没有办法在使用 filter 后返回两个 DStream？例如，当我过滤一个DStream时，过滤后的将存储在一个DStream中，未过滤的将存储在另一个DStream中。

Answer 1

如果它是内置的，这可以更有效地完成，但是

def partition[A](stream: DStream[A])(pred: A => Boolean) {
  val stream1 = stream.map(x => (x, pred(x)).cache()
  val good = stream1.filter(_._2).map(_._1)
  val bad = stream1.filter(!_._2).map(_._1)
  (good, bad)
}

注意 cache() 是为了确保 stream1 只被计算一次；如果 pred 足够简单，并且 stream 已经缓存，只需 (stream.filter(pred), stream.filter(x => !pred(x) )) 应该更快。