scala - Spark : How do I pass a PartialFunction to a DStream?

Question

我正在尝试将部分函数传递给通过滑动窗口在 DStream 批处理中捕获的所有 RDD 的并集。假设我在离散为 1 秒批处理的流上构造了一个超过 10 秒的窗口操作:

val ssc = new StreamingContext(new SparkConf(), Seconds(1))
val stream = ssc.socketStream(...)
val window = stream.window(Seconds(10))

我的 window 将有 K 个 RDD。我想在所有 K 个 RDD 的并集上使用 collect(f: PartialFunction[T, U])。我可以使用 foreachRDD 调用联合运算符 ++，但我想返回一个 RDD 而不是 Unit 并且避免副作用。

我要找的是reducer之类的

def reduce(f: (RDD[T], RDD[T]) ⇒ RDD[T]): RDD[T]

在我可以这样使用的 DStream 上:

window.reduce(_ ++ _).transform(_.collect(myPartialFunc))

但这在 Spark Streaming API 中是不可用的。

有没有人有什么好主意可以将流中捕获的 RDD 组合成单个 RDD，以便我可以传递部分函数？还是为了实现我自己的 RDD reducer？也许此功能会出现在后续的 Spark 版本中？

Answer 1

DStream 操作不直接支持部分功能，但实现相同功能并不难。

例如，让我们采用一个简单的部分函数，​​它接受一个字符串，如果它是一个数字，则生成一个字符串的 Int:

val pf:PartialFunction[String,Int] = {case x if (Try(x.toInt).isSuccess) => x.toInt}

我们有一个字符串数据流:

val stringDStream:DStream[String] = ??? // use your stream source here

然后我们可以像这样将偏函数应用于 DStream:

val intDStream = stringDStream.filter(x => pf.isDefinedAt(x)).map(pf)