Spark 流式传输：跨批次缓存 DStream 结果

Question

使用 Spark 流式传输 (1.6) 我有一个文件流，用于读取批量大小为 2 秒的查找数据，但是文件仅每小时复制到目录中。
一旦有一个新文件，它的内容就会被流读取，这就是我想要缓存到内存中并保留在那里的内容 直到读取新文件。
我想将这个数据集加入到另一个流中，因此我想缓存。

这是Batch lookup data for Spark streaming的后续问题。
答案确实适用于updateStateByKey，但是我不知道如何处理 KV 对的情况 从查找文件中删除，因为updateStateByKey 中的值序列不断增长。 此外，任何提示如何使用 mapWithState 来做到这一点都会很棒。

这是我到目前为止尝试过的，但数据似乎没有被持久化：

val dictionaryStream = ssc.textFileStream("/my/dir")
dictionaryStream.foreachRDD{x => 
  if (!x.partitions.isEmpty) {
    x.unpersist(true)
    x.persist()
  }
}

Answer 1

DStreams 可以直接使用 persist 方法持久化，该方法持久化流中的每个 RDD：

dictionaryStream.persist

根据the official documentation这个自动申请

reduceByWindow 和reduceByKeyAndWindow 等基于窗口的操作以及updateStateByKey 等基于状态的操作

所以在你的情况下应该不需要显式缓存。也不需要手动取消持久化。再次引用the docs：

默认情况下，DStream 转换生成的所有输入数据和持久化 RDD 都会自动清除

并根据管道中使用的转换自动调整保留期。

关于mapWithState，您必须提供StateSpec。一个最小的示例需要一个函数，该函数采用当前value 和先前状态的key、Option。假设您有DStream[(String, Long)]，并且您想记录迄今为止的最大值：

val state = StateSpec.function(
  (key: String, current: Option[Double], state: State[Double]) => {
    val max  = Math.max(
      current.getOrElse(Double.MinValue),
      state.getOption.getOrElse(Double.MinValue)
    )
    state.update(max)
    (key, max)
  }
)

val inputStream: DStream[(String, Double)] = ??? 
inputStream.mapWithState(state).print()

还可以提供初始状态、超时间隔和捕获当前批处理时间。最后两个可以用来对一段时间没有更新的keys实现删除策略。