积累 Spark Streaming DStream 的最佳方案答案

【问题标题】：Best solution to accumulate Spark Streaming DStream积累 Spark Streaming DStream 的最佳方案
【发布时间】：2016-12-22 01:01:11
【问题描述】：

我正在寻找在 Spark DStream 中累积最后 N 条消息的最佳解决方案。我还想指定要保留的消息数。

例如，给定以下流，我想保留最后 3 个元素：

Iteration  New message  Downstream
1          A            [A]
2          B            [A, B]
3          C            [A, B, C]
4          D            [B, C, D]

到目前为止，我正在研究 DStream 上的以下方法：

updateStateByKey：鉴于所有消息都具有相同的密钥，我可以这样做。但看起来有点奇怪，为什么这需要知道密钥。
mapWithState：Scala 中的 API 对于这么简单的事情来说太乏味了
window：似乎没有做这项工作，它还需要一个时间值来进行窗口化而不是最后一个元素数
累加器：尚未真正使用Accumulators in Spark docs

实现这一目标的最佳解决方案是什么？

【问题讨论】：

标签： scala apache-spark spark-streaming dstream

【解决方案1】：

mapWithState正是你所需要的，而且绝对不会太繁琐：

case class Message(x: String)
def statefulTransformation(key: Int,
                           value: Option[Message],
                           state: State[mutable.MutableList[Message]]): Option[Message] = {
  def updateState(value: Message): Message = {
    val updatedList =
      state
        .getOption()
        .map(list => if (list.size > 3) list.drop(1) :+ value else list :+ value)
      .getOrElse(mutable.MutableList(value))

    state.update(updatedList)
    value
  }

  value.map(updateState)
}

现在你只需要：

val stateSpec = StateSpec.function(statefulTransformation _)
dStream.mapWithState(stateSpec)

旁注 - 我使用mutable.MutableList 来表示恒定时间追加。

【讨论】：