并行处理 DStream 中的 RDD

Question

我遇到了以下在 Spark Streaming 中处理消息的代码：

val listRDD = ssc.socketTextStream(host, port)
listRDD.foreachRDD(rdd => {
  rdd.foreachPartition(partition => {
    // Should I start a separate thread for each RDD and/or Partition?
    partition.foreach(message => {
      Processor.processMessage(message)
    })
  })
})

这对我有用，但我不确定这是否是最好的方法。我知道 DStream 由“一对多”的 RDD 组成，但是这段代码一个接一个地依次处理 RDD，对吗？难道没有更好的方法 - 一种方法或函数 - 我可以使用以便并行处理 DStream 中的所有 RDD 吗？我应该为每个 RDD 和/或分区启动一个单独的线程吗？我是否误解了这段代码在 Spark 下的工作方式？

不知何故，我认为这段代码没有利用 Spark 中的并行性。

Answer 1

为了方便和高效，流被划分为小的 RDD（请查看 micro-batching。但您确实不需要将每个 RDD 划分为分区，甚至不需要将流划分为 RDD。

这完全取决于Processor.processMessage 的真正含义。如果它是单个转换函数，您只需执行 listRDD.map(Processor.processMessage) 即可获得处理消息的任何结果的流，并行计算，无需您做很多其他事情。

如果Processor 是一个保持状态的可变对象（例如，计算消息的数量），那么事情就更复杂了，因为您需要定义许多这样的对象来解释并行性，并且还需要以某种方式合并结果稍后。