065 updateStateByKey的函数API

一：使用场景

1.应用场景

　　数据的累加

　　一段时间内的数据的累加

2.说明

　　每个批次都输出自己批次的数据，

　　这个时候，可以使用这个API，使得他们之间产生联系。

3.说明2

　　在累加器的时候，起到的效果和这里的说明想法有些相同，都可以输出上一个批次的信息

二：程序

1.需要启动一些服务

　　需要使用hadoop

　　

2.程序

 package com.stream.it

 import kafka.serializer.StringDecoder

 import org.apache.spark.rdd.RDD

 import org.apache.spark.storage.StorageLevel

 import org.apache.spark.streaming.kafka.KafkaUtils

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

 import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

 object UpdateStateByKeyKafkaWordcount {

   def main(args: Array[String]): Unit = {

     val conf=new SparkConf()

         .setAppName("spark-streaming-wordcount")

           .setMaster("local[*]")

     val sc=SparkContext.getOrCreate(conf)

     val ssc=new StreamingContext(sc,Seconds(15))

     val kafkaParams=Map("group.id"->"stream-sparking-0",

           "zookeeper.connect"->"linux-hadoop01.ibeifeng.com:2181/kafka",

           "auto.offset.reset"->"smallest"

     )

     val topics=Map("beifeng"->1)

     val dStream=KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](

       ssc,             //给定sparkStreaming的上下文

       kafkaParams,     //kafka的参数信息，通过kafka HightLevelComsumerApi连接

       topics,          //给定读取对应的topic的名称以及读取数据的线程数量

       StorageLevel.MEMORY_AND_DISK_2     //数据接收器接收到kafka的数据后的保存级别

     ).map(_._2)

     // 当调用updateStateByKey函数API的时候，必须给定checkpoint dir

     // 路径对应的文件夹不能存在

     ssc.checkpoint("hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/chkdir01")

     /**

       def updateStateByKey[S: ClassTag](

         updateFunc: (Seq[V], Option[S]) => Option[S],

         partitioner: Partitioner,

         initialRDD: RDD[(K, S)]

       ): DStream[(K, S)]

       */

     val resultWordcount=dStream

       .filter(line=>line.nonEmpty)

         .flatMap(line=>line.split(" ").map((_,1)))

         .reduceByKey(_+_)

         .updateStateByKey(

           (values: Seq[Int], state: Option[Long]) => {

             // 从value中获取累加值

             val sum = values.sum

             // 获取以前的累加值

             val oldStateSum = state.getOrElse(0L)

             // 更新状态值并返回

             Some(oldStateSum + sum)

           }

       )

     resultWordcount.foreachRDD(rdd=>{

       rdd.foreachPartition(iter=>iter.foreach(println))

     })

     //启动

     ssc.start()

     //等到

     ssc.awaitTermination()

   }

 }

三：updateStateByKey的优化

1.说明

　　主要的情况是，程序停止，刚刚累加的数据不再存在。

　　重启后效果如下：

　　　　只剩下，已经被checkPoint的数据，后面的数据不再存在。

　　　　

2.优化的程序

　　多加两个参数。

 package com.stream.it

 import kafka.serializer.StringDecoder

 import org.apache.spark.rdd.RDD

 import org.apache.spark.storage.StorageLevel

 import org.apache.spark.streaming.kafka.KafkaUtils

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

 import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

 object UpdateStateByKeyKafkaWordcount {

   def main(args: Array[String]): Unit = {

     val conf=new SparkConf()

         .setAppName("spark-streaming-wordcount")

           .setMaster("local[*]")

     val sc=SparkContext.getOrCreate(conf)

     val ssc=new StreamingContext(sc,Seconds(15))

     val kafkaParams=Map("group.id"->"stream-sparking-0",

           "zookeeper.connect"->"linux-hadoop01.ibeifeng.com:2181/kafka",

           "auto.offset.reset"->"largest"

     )

     val topics=Map("beifeng"->1)

     val dStream=KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](

       ssc,             //给定sparkStreaming的上下文

       kafkaParams,     //kafka的参数信息，通过kafka HightLevelComsumerApi连接

       topics,          //给定读取对应的topic的名称以及读取数据的线程数量

       StorageLevel.MEMORY_AND_DISK_2     //数据接收器接收到kafka的数据后的保存级别

     ).map(_._2)

     // 当调用updateStateByKey函数API的时候，必须给定checkpoint dir

     // 路径对应的文件夹不能存在

     ssc.checkpoint("hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/chkdir01")

     // 初始化updateStateByKey用到的状态值

     // 从保存状态值的地方(HBase)读取状态值， 这里采用模拟的方式

     val initialRDD: RDD[(String, Long)] = sc.parallelize(

       Array(

         ("hadoop", 100L),

         ("spark", 25L)

       )

     )

     /**

       def updateStateByKey[S: ClassTag](

         updateFunc: (Seq[V], Option[S]) => Option[S],

         partitioner: Partitioner,

         initialRDD: RDD[(K, S)]

       ): DStream[(K, S)]

       */

     val resultWordcount=dStream

       .filter(line=>line.nonEmpty)

         .flatMap(line=>line.split(" ").map((_,1)))

         .reduceByKey(_+_)

         .updateStateByKey(

           (values: Seq[Int], state: Option[Long]) => {

             // 从value中获取累加值

             val sum = values.sum

             // 获取以前的累加值

             val oldStateSum = state.getOrElse(0L)

             // 更新状态值并返回

             Some(oldStateSum + sum)

           },

           new HashPartitioner(ssc.sparkContext.defaultParallelism), // 分区器

           initialRDD // 初始化状态值

       )

     resultWordcount.foreachRDD(rdd=>{

       rdd.foreachPartition(iter=>iter.foreach(println))

     })

     //启动

     ssc.start()

     //等到

     ssc.awaitTermination()

   }

 }

3.运行效果

　　

4.注意点

　　需要有checkPoint的路径。

　　累加值存在硬盘中，长时间不访问会被删除。