如何在Spark Streaming中构建查找映射？

Question

在Spark Streaming应用程序中维护应用程序状态的最佳方法是什么？

我知道两种方法：

1. 使用“联合”（Union）操作附加到查找RDD，并在每次合并后保留它。
2. 将状态保存在文件或数据库中，并在每个批处理的开始时加载它。

我的问题是从性能角度来看哪个更好？另外，还有更好的方法吗？

Answer 1

您应该真正使用mapWithState（spec：StateSpec [K，V，StateType，MappedType]）如下：

import org.apache.spark.streaming.{ StreamingContext, Seconds }
val ssc = new StreamingContext(sc, batchDuration = Seconds(5))

// checkpointing is mandatory
ssc.checkpoint("_checkpoints")

val rdd = sc.parallelize(0 to 9).map(n => (n, n % 2 toString))
import org.apache.spark.streaming.dstream.ConstantInputDStream
val sessions = new ConstantInputDStream(ssc, rdd)

import org.apache.spark.streaming.{State, StateSpec, Time}
val updateState = (batchTime: Time, key: Int, value: Option[String], state: State[Int]) => {
  println(s">>> batchTime = $batchTime")
  println(s">>> key       = $key")
  println(s">>> value     = $value")
  println(s">>> state     = $state")
  val sum = value.getOrElse("").size + state.getOption.getOrElse(0)
  state.update(sum)
  Some((key, value, sum)) // mapped value
}
val spec = StateSpec.function(updateState)
val mappedStatefulStream = sessions.mapWithState(spec)

mappedStatefulStream.print()