Spark Streaming updateStateByKey案例实战和内幕源码解密

本节课程主要分二个部分：

1、Spark Streaming updateStateByKey案例实战
2、Spark Streaming updateStateByKey源码解密

第一部分：

updateStateByKey的主要功能是随着时间的流逝，在Spark Streaming中能够为每个能够经过CheckPoint来维护一份state状态，经过更新函数对该key的状态不断更新；对每个新批次的数据（batch）而言，Spark Streaming经过使用updateStateByKey为已经存在的key进行state的状态更新（对每一个新出现的key，会一样执行state的更新函数操做）；可是若是经过更新函数对state更新后返回none的话，此时刻key对应的state状态被删除掉，须要特别说明的是state能够是任意类型的数据结构，这就为咱们的计算带来无限的想象空间；

很是重要：

若是要不断的更新每一个key的state，就必定会涉及到状态的保存和容错，这个时候就须要开启checkpoint机制和功能，须要说明的是checkpoint的数据能够保存一些存储在文件系统上的内容，例如：程序未处理的但已经拥有状态的数据。

补充说明：

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

案例实战源码：

1.编写源码：

ackage org.apache.spark.examples.streaming;

import java.util.Arrays;

import java.util.List;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import org.apache.spark.streaming.Durations;

import org.apache.spark.streaming.api.java.JavaDStream;

import org.apache.spark.streaming.api.java.JavaPairDStream;

import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;

import org.apache.spark.streaming.api.java.JavaStreamingContext;

import com.google.common.base.Optional;

import scala.Tuple2;

 

public class UpdateStateByKeyDemo {

public static void main(String[] args) {

/*

* 第一步：配置SparkConf：

* 1，至少2条线程：由于Spark Streaming应用程序在运行的时候，至少有一条

* 线程用于不断的循环接收数据，而且至少有一条线程用于处理接受的数据（不然的话没法

* 有线程用于处理数据，随着时间的推移，内存和磁盘都会不堪重负）；

* 2，对于集群而言，每一个Executor通常确定不止一个Thread，那对于处理Spark Streaming的

* 应用程序而言，每一个Executor通常分配多少Core比较合适？根据咱们过去的经验，5个左右的

* Core是最佳的（一个段子分配为奇数个Core表现最佳，例如3个、5个、7个Core等）；

*/

SparkConf conf = new SparkConf().setMaster("local[2]").

setAppName("UpdateStateByKeyDemo");

/*

* 第二步：建立SparkStreamingContext：

* 1，这个是SparkStreaming应用程序全部功能的起始点和程序调度的核心

* SparkStreamingContext的构建能够基于SparkConf参数，也可基于持久化的SparkStreamingContext的内容

* 来恢复过来（典型的场景是Driver崩溃后从新启动，因为Spark Streaming具备连续7*24小时不间断运行的特征，

* 全部须要在Driver从新启动后继续上衣系的状态，此时的状态恢复须要基于曾经的Checkpoint）；

* 2，在一个Spark Streaming应用程序中能够建立若干个SparkStreamingContext对象，使用下一个SparkStreamingContext

* 以前须要把前面正在运行的SparkStreamingContext对象关闭掉，由此，咱们得到一个重大的启发SparkStreaming框架也只是

* Spark Core上的一个应用程序而已，只不过Spark Streaming框架箱运行的话须要Spark工程师写业务逻辑处理代码；

*/

JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

//报错解决办法作checkpoint,开启checkpoint机制，把checkpoint中的数据放在这里设置的目录中，

//生产环境下通常放在HDFS中

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

/*

* 第三步：建立Spark Streaming输入数据来源input Stream：

* 1，数据输入来源能够基于File、HDFS、Flume、Kafka、Socket等

* 2, 在这里咱们指定数据来源于网络Socket端口，Spark Streaming链接上该端口并在运行的时候一直监听该端口

* 的数据（固然该端口服务首先必须存在）,而且在后续会根据业务须要不断的有数据产生(固然对于Spark Streaming

* 应用程序的运行而言，有无数据其处理流程都是同样的)； 

* 3,若是常常在每间隔5秒钟没有数据的话不断的启动空的Job实际上是会形成调度资源的浪费，由于并无数据须要发生计算，因此

* 实例的企业级生成环境的代码在具体提交Job前会判断是否有数据，若是没有的话就再也不提交Job；

*/

JavaReceiverInputDStream lines = jsc.socketTextStream("hadoop100", 9999);

/*

* 第四步：接下来就像对于RDD编程同样基于DStream进行编程！！！缘由是DStream是RDD产生的模板（或者说类），在Spark Streaming具体

* 发生计算前，其实质是把每一个Batch的DStream的操做翻译成为对RDD的操做！！！

*对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算

    * 第4.1步：讲每一行的字符串拆分红单个的单词

    */

JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { //若是是Scala，因为SAM转换，因此能够写成val words = lines.flatMap { line => line.split(" ")}

@Override

public Iterable<String> call(String line) throws Exception {

return Arrays.asList(line.split(" "));

}

});

/*

      * 第四步：对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算

      * 第4.2步：在单词拆分的基础上对每一个单词实例计数为1，也就是word => (word, 1)

      */

JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

@Override

public Tuple2<String, Integer> call(String word) throws Exception {

return new Tuple2<String, Integer>(word, 1);

}

});

/*

      * 第四步：对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算

      *第4.3步：在这里是经过updateStateByKey来以Batch Interval为单位来对历史状态进行更新，

      * 这是功能上的一个很是大的改进，不然的话须要完成一样的目的，就可能须要把数据保存在Redis、

      * Tagyon或者HDFS或者HBase或者数据库中来不断的完成一样一个key的State更新，若是你对性能有极为苛刻的要求，

      * 且数据量特别大的话，能够考虑把数据放在分布式的Redis或者Tachyon内存文件系统中；

      * 固然从Spark1.6.x开始能够尝试使用mapWithState，Spark2.X后mapWithState应该很是稳定了。

 */

JavaPairDStream<String, Integer> wordsCount = pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { //对相同的Key，进行Value的累计（包括Local和Reducer级别同时Reduce）

@Override

public Optional<Integer> call(List<Integer> values, Optional<Integer> state)

throws Exception {

Integer updatedValue = 0 ;

if(state.isPresent()){

updatedValue = state.get();

}

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请求：

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
        }
      }
    }
  }

总结：

姜伟

备注：93课

更多私密内容，请关注微信公众号：DT_Spark