Spark Streaming 流计算状态管理

    spark流计算的数据是以窗口的形式，源源不断的流过来的。如果每个窗口之间的数据都有联系的话，那么就需要对前一个窗口的数据做状态管理。spark有提供了两种模型来达到这样的功能，一个是updateStateByKey，另一个是mapWithState ，后者属于Spark1.6之后的版本特性，性能是前者的数十倍。
    updateStateByKey

    通过源码查看发现，这个模型的核心思想就是将之前有状态的RDD和当前的RDD做一次cogroup，得到一个新的状态的RDD，以此迭代。updateStateByKey函数在DStream以及MappedDStream中是没有的，后来发现DStrem的伴生对象有一个隐式转换函数toPairDStreamFunctions可以将DStream转换成PairDStreamFunction。object DStream extends scala.AnyRef with scala.Serializable {
implicit def toPairDStreamFunctions[K, V](stream : org.apache.spark.streaming.dstream.DStream[scala.Tuple2[K, V]])(implicit kt : scala.reflect.ClassTag[K], vt : scala.reflect.ClassTag[V], ord : scala.Ordering[K] = { /* compiled code */ }) : org.apache.spark.streaming.dstream.PairDStreamFunctions[K, V] = { /* compiled code */ }
private[streaming] def getCreationSite() : org.apache.spark.util.CallSite = { /* compiled code */ }
}    PairDStreamFunctions中存在updateStateByKey函数，源码如下，传入具体的updateFunc函数，此函数需要传入当前的key对应的所以值，以及当前key的状态。具体状态更新函数体可以根据业务具体实现。 def updateStateByKey[S](updateFunc : scala.Function2[scala.Seq[V], scala.Option[S], scala.Option[S]])(implicit evidence$4 : scala.reflect.ClassTag[S]) : org.apache.spark.streaming.dstream.DStream[scala.Tuple2[K, S]] = { /* compiled code */ }    下面的案例代码，就是读取kafka streaming数据，每个窗口根据性别对分数进行求和。然后通过updateStateByKey更新之前的状态，达到对所有流过的数据求和的效果。    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark-streaming")
val ssc = new StreamingContext(sparkConf, Seconds(10))
ssc.checkpoint("hdfs://bigdata05:9000/spark/streaming/cyony")
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "bigdata05:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "spark-streaming-05",
"auto.offset.reset" -> "earliest",
"enable.auto.commit" -> (true: java.lang.Boolean)
)
val messages = KafkaUtils.createDirectStream[String, String](
ssc, PreferConsistent, Subscribe[String, String](Set("cyony"), kafkaParams))

val updateFun = (currentValue: Seq[Int], preValue: Option[Int]) => {
Some(currentValue.sum + preValue.getOrElse(0))
}

messages.map(_.value()).map(JSON.parseFull(_).get.asInstanceOf[Map[String, String]])
.map(map => (map.get("sex").get.toInt, map.get("score").get.toInt)).reduceByKey(_ + _)
.updateStateByKey(updateFun)
.print()

ssc.start()
ssc.awaitTermination()    样例数据：

    {"name":"cyony1","score":"90","sex":"1"}

    {"name":"cyony2","score":"76","sex":"0"}
    这种模型，每次窗口触发，都会将两个RDD执行cogroup操作，非常的耗时，所以spark在1.6以后的版本提供了新的流状态管理方式。

    mapWithState    这个模型定义了一种新的RDD叫MapWithStateRDD，这个RDD只能存放MapWithStateRDDRecord元素。此元素存放了一个分区的所有Key的状态，以及计算结果。这样每次只要更新这个Record，不需要重新生成RDD，同样保持了RDD的不变性。源码要求的传入函数接口如下，接口要求传入一个StateSpec函数。具体的实现如下面的案例。 def mapWithState[StateType, MappedType](spec : org.apache.spark.streaming.StateSpec[K, V, StateType, MappedType])(implicit evidence$2 : scala.reflect.ClassTag[StateType], evidence$3 : scala.reflect.ClassTag[MappedType]) : org.apache.spark.streaming.dstream.MapWithStateDStream[K, V, StateType, MappedType] = { /* compiled code */ }
    实现上面同样的功能，用MapWithState方式实现代码如下：    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark-streaming")
val ssc = new StreamingContext(sparkConf, Seconds(10))
ssc.checkpoint("hdfs://bigdata05:9000/spark/streaming/cyony")
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "bigdata05:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "spark-streaming-05",
"auto.offset.reset" -> "earliest",
"enable.auto.commit" -> (true: java.lang.Boolean)
)
val messages = KafkaUtils.createDirectStream[String, String](
ssc, PreferConsistent, Subscribe[String, String](Set("cyony"), kafkaParams))

val mappingFun = (sex: Int, score: Option[Int], state: State[Int]) => {
val sum = score.getOrElse(0) + state.getOption().getOrElse(0)
state.update(sum)
(sex, sum)
}

messages.map(_.value()).map(JSON.parseFull(_).get.asInstanceOf[Map[String, String]])
.map(map => (map.get("sex").get.toInt, map.get("score").get.toInt)).reduceByKey(_ + _)
.mapWithState(StateSpec.function(mappingFun))
.print()

ssc.start()
ssc.awaitTermination()
    通过以上两种方式实际运行对比可以发现，如果当前窗口期没有新的数据过来，mapstate方式是根本不会触发状态更新操作的，但是updateState方式就会触发更新操作。这个和他的模型原理有关，进一步佐证了updateState方式会每次都执行cogroup操作RDD，生成新的RDD。
    以上代码运行，maven pom文件依赖如下
        <dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.2.0.cloudera1</version>
</dependency>

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.2.0.cloudera1</version>
</dependency>

<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-actor_2.12</artifactId>
<version>2.5.4</version>
</dependency>

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10-assembly_2.11</artifactId>
<version>2.1.0</version>
</dependency>

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.2.0</version>
</dependency>