第93课：SparkStreaming updateStateByKey 基本操作综合案例实战和内幕源码解密

   Spark Streaming的DStream为我们提供了一个updateStateByKey方法，它的主要功能是可以随着时间的流逝在Spark Streaming中为每一个key维护一份state状态，通过更新函数对该key的状态不断更新。对每一个新的batch而言，Spark Streaming会在使用updateStateByKey的时候为已经存在的key进行state的状态更新（对每个新出现的key，会同样执行state的更新函数操作），但是如果通过更新函数对state更新后返回none的话，此时刻key对应的state状态被删除掉，需要特别说明的是state可以是任意类型的数据结构，这就为我们的计算带来无限的想象空间；

  重点来了！！！如果要不断的更新每个key的state，就一定会涉及到状态的保存和容错，这个时候就需要开启checkpoint机制和功能，需要说明的是checkpoint可以保存一切可以存储在文件系统上的内容，例如：程序未处理的数据及已经拥有的状态。

  补充说明：关于流式处理对历史状态进行保存和更新具有重大实用意义，例如进行广告（投放广告和运营广告效果评估的价值意义，热点随时追踪、热力图）

  简单的来说，如果我们需要进行wordcount，每个batchInterval都会计算出新的一批数据，这批数据如何更新到以前计算的结果上？updateStateByKey就能实现此功能。

函数定义如下：

def updateStateByKey[S: ClassTag](
    updateFunc: (Seq[V], Option[S]) => Option[S]
  ): DStream[(K, S)] = ssc.withScope {
  updateStateByKey(updateFunc, defaultPartitioner())
}

updateStateByKey 需要传入一个函数，该函数有两个参数Seq[V]表示最新一次reduce的值的序列，Option[s]表示的是key对应的以前的值。返回的时一个key的最新值。

下面我们用实例演示：

package com.dt.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Administrator on 2016/5/3.
 */
object UpdateStateByKeyDemo {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("UpdateStateByKeyDemo")
    val ssc = new StreamingContext(conf,Seconds(20))
    //要使用updateStateByKey方法，必须设置Checkpoint。
    ssc.checkpoint("/checkpoint/")
    val socketLines = ssc.socketTextStream("spark-master",9999)

    socketLines.flatMap(_.split(",")).map(word=>(word,1))
      .updateStateByKey(
        (currValues:Seq[Int],preValue:Option[Int]) =>{
       val currValue = currValues.sum
         Some(currValue + preValue.getOrElse(0))
    }).print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}

打包上传至spark集群。

打开nc，发送测试数据

root@spark-master:~# nc -lk 9999
hadoop,spark,scala,hive
hadoop,Hbase,spark

运行spark 程序

root@spark-master:~# /usr/local/spark/spark-1.6.0-bin-hadoop2.6/bin/spark-submit --class com.dt.spark.streaming.UpdateStateByKeyDemo  --master spark://spark-master:7077 ./spark.jar

查看运行结果：

-------------------------------------------
Time: 1462282180000 ms
-------------------------------------------
(scala,1)
(hive,1)
(spark,2)
(hadoop,2)
(Hbase,1)

我们在nc中再输入一些数据

root@spark-master:~# nc -lk 9999
hadoop,spark,scala,hive
hadoop,Hbase,spark
hadoop,spark,scala,hive
hadoop,Hbase,spark

再次查看结果：

-------------------------------------------
Time: 1462282200000 ms
-------------------------------------------
(scala,2)
(hive,2)
(spark,4)
(hadoop,4)
(Hbase,2)

可见，它将我们两次统计结果合并了。

备注：

1、DT大数据梦工厂微信公众号DT_Spark 
2、IMF晚8点大数据实战YY直播频道号：68917580
3、新浪微博: http://www.weibo.com/ilovepains