第93讲：SparkStreamingupdateStateByKey案例实战和内幕源码

throws Exception {

//第一个参数就是key传进来的数据，第二个参数是曾经已有的数据

Integer updatedValue = 0 ;//如果第一次，state没有，updatedValue为0，如果有，就获取

if(state.isPresent()){

updatedValue = state.get();

}

//遍历batch传进来的数据可以一直加，随着时间的流式会不断去累加相同key的value的结果。

for(Integer value: values){

updatedValue += value;

}

return Optional.of(updatedValue);//返回更新的值

}

});

/*

*此处的print并不会直接出发Job的执行，因为现在的一切都是在Spark Streaming框架的控制之下的，对于Spark Streaming

*而言具体是否触发真正的Job运行是基于设置的Duration时间间隔的

*诸位一定要注意的是Spark Streaming应用程序要想执行具体的Job，对Dtream就必须有output Stream操作，

*output Stream有很多类型的函数触发，类print、saveAsTextFile、saveAsHadoopFiles等，最为重要的一个

*方法是foraeachRDD,因为Spark Streaming处理的结果一般都会放在Redis、DB、DashBoard等上面，foreachRDD

*主要就是用用来完成这些功能的，而且可以随意的自定义具体数据到底放在哪里！！！

*/

wordsCount.print();

/*

* Spark Streaming执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于

* 接受应用程序本身或者Executor中的消息；

*/

jsc.start();

jsc.awaitTermination();

jsc.close();

}

2.创建checkpoint目录：

jsc.checkpoint("/usr/local/tmp/checkpoint");

3.　在eclipse中通过run 方法启动main函数：

4.启动hdfs服务并发送nc -lk 9999请求：

继续输入Hello hdfs

结果就是累加后的结果，

(Hello,3)

(SPark,1)

(hdfs,2)

再次输入Hello SPark，下一个batch就会继续累加，(Hello,3)(SPark,2) (hdfs,2)的结果输出

5.查看checkpoint目录输出：因为是二进制

源码解析：

1.PairDStreamFunctions类：

/**

 * Return a new "state" DStream where the state for each key is updated by applying
 * the given function on the previous state of the key and the new values of each key.
 * Hash partitioning is used to generate the RDDs with Spark's default number of partitions.
 * @param updateFunc State update function. If `this` function returns None, then
 *                   corresponding state key-value pair will be eliminated.
 * @tparam S State type
 */
def updateStateByKey[S: ClassTag](
    updateFunc: (Seq[V], Option[S]) => Option[S]
  ): DStream[(K, S)] = ssc.withScope {
  updateStateByKey(updateFunc, defaultPartitioner())
}

/**
 * Return a new "state" DStream where the state for each key is updated by applying
 * the given function on the previous state of the key and the new values of the key.
 * org.apache.spark.Partitioner is used to control the partitioning of each RDD.
 * @param updateFunc State update function. If `this` function returns None, then
 *                   corresponding state key-value pair will be eliminated.
 * @param partitioner Partitioner for controlling the partitioning of each RDD in the new
 *                    DStream.
 * @tparam S State type
 */
def updateStateByKey[S: ClassTag](
    updateFunc: (Seq[V], Option[S]) => Option[S],
    partitioner: Partitioner
  ): DStream[(K, S)] = ssc.withScope {
  val cleanedUpdateF = sparkContext.clean(updateFunc)
  val newUpdateFunc = (iterator: Iterator[(K, Seq[V], Option[S])]) => {
    iterator.flatMap(t => cleanedUpdateF(t._2, t._3).map(s => (t._1, s)))
  }
  updateStateByKey(newUpdateFunc, partitioner, true)
}

/**
 * Return a new "state" DStream where the state for each key is updated by applying
 * the given function on the previous state of the key and the new values of each key.
 * org.apache.spark.Partitioner is used to control the partitioning of each RDD.
 * @param updateFunc State update function. Note, that this function may generate a different
 *                   tuple with a different key than the input key. Therefore keys may be removed
 *                   or added in this way. It is up to the developer to decide whether to
 *                   remember the partitioner despite the key being changed.
 * @param partitioner Partitioner for controlling the partitioning of each RDD in the new
 *                    DStream
 * @param rememberPartitioner Whether to remember the paritioner object in the generated RDDs.
 * @tparam S State type
 */

def updateStateByKey[S: ClassTag](
    updateFunc: (Iterator[(K, Seq[V], Option[S])]) => Iterator[(K, S)],
    partitioner: Partitioner,
    rememberPartitioner: Boolean
  ): DStream[(K, S)] = ssc.withScope {
   new StateDStream(self, ssc.sc.clean(updateFunc), partitioner, rememberPartitioner, None)
}

override def compute(validTime: Time): Option[RDD[(K, S)]] = {

  // Try to get the previous state RDD
  getOrCompute(validTime - slideDuration) match {

    case Some(prevStateRDD) => {    // If previous state RDD exists

      // Try to get the parent RDD
      parent.getOrCompute(validTime) match {
        case Some(parentRDD) => {   // If parent RDD exists, then compute as usual
          computeUsingPreviousRDD (parentRDD, prevStateRDD)
        }
        case None => {    // If parent RDD does not exist

          // Re-apply the update function to the old state RDD
          val updateFuncLocal = updateFunc
          val finalFunc = (iterator: Iterator[(K, S)]) => {
            val i = iterator.map(t => (t._1, Seq[V](), Option(t._2)))
            updateFuncLocal(i)
          }
          val stateRDD = prevStateRDD.mapPartitions(finalFunc, preservePartitioning)
          Some(stateRDD)
        }
      }
    }

    case None => {    // If previous session RDD does not exist (first input data)

      // Try to get the parent RDD
      parent.getOrCompute(validTime) match {
        case Some(parentRDD) => {   // If parent RDD exists, then compute as usual
          initialRDD match {
            case None => {
              // Define the function for the mapPartition operation on grouped RDD;
              // first map the grouped tuple to tuples of required type,
              // and then apply the update function
              val updateFuncLocal = updateFunc
              val finalFunc = (iterator : Iterator[(K, Iterable[V])]) => {
                updateFuncLocal (iterator.map (tuple => (tuple._1, tuple._2.toSeq, None)))
              }

              val groupedRDD = parentRDD.groupByKey (partitioner)
              val sessionRDD = groupedRDD.mapPartitions (finalFunc, preservePartitioning)
              // logDebug("Generating state RDD for time " + validTime + " (first)")
              Some (sessionRDD)
            }
            case Some (initialStateRDD) => {
              computeUsingPreviousRDD(parentRDD, initialStateRDD)
            }
          }
        }
        case None => { // If parent RDD does not exist, then nothing to do!
          // logDebug("Not generating state RDD (no previous state, no parent)")
          None
        }
      }
    }
  }