Spark Streaming中的微批合并

Question

那么我需要按照以下方式处理这些消息：

1. 在每个消息(a,b)->x上应用一个（couple of）函数，输出(a,b)->y
2. 应在消息ab->seq[y]上应用一个函数，其中ab={（a,b）for all b in b}

（稍后有一个传递，需要“转置”消息以跨所有a处理，但这不是这里的问题）

我如何实现这样的消息合并，从步骤1到步骤2？

它看起来像是子键a上的groupby，但据我所知，方法groupby将在每个微批处理中应用。我需要的是，对于每个a，等待接收到所有B（假设一个简单的计数系统可以工作）。再一次抛开输入数据中的丢失b和错误。

在没有知识的情况下，我会尝试看看这样的合并是否可以通过添加到hdfs文件来实现，每个文件对应一个。并尝试在这些文件满后触发第二个流进程。即。当它包含所有b时，将该文件移动到一个输入目录进行步骤2。但是：

1. 我不知道这样的附加是否可以在HDFS上实现
2. 两个sparkStreamingContext需要并行运行，每个步骤一个。这看起来是个问题（？）。
3. 我知道通过hdfs会破坏spark（流）的“精确一次”属性

Answer 1

您可以创建主RDD，并使用RDD.union将流生成的微RDD合并到主RDD。类似于：

var masterRDD: RDD[(Long,Long), String] = sc.emptyRDD  // guessing on RDD type

myStream.foreachRDD(rdd => {
  if (! rdd.isEmpty) {
    masterRDD.union(rdd)

    masterRDD.groupBy(...).....
  }
})

您也应该花点时间阅读检查点，特别是：

数据检查点-将生成的RDD保存到可靠的存储区。在一些跨多批组合数据的有状态转换中，这是必需的。在这样的转换中，生成的RDD依赖于先前批次的RDD，这导致依赖链的长度随着时间不断增加。为了避免恢复时间的这种无限增长（与依赖链成比例），有状态转换的中间RDD被周期性地检查指向可靠存储（例如HDF)，以切断依赖链。