067 HA与updateStateByKey结合

　　是HA与updateStateByKey相结合的程序。

　　有点问题，有点奇怪，重启项目后运行没有问题，但是第三次启动的时候，就不会在打印数据了，有点问题。

1.程序

package com.stream.it
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 通过HA管理KAFKA的offset偏移量
  * Created by ibf on 03/04.
  */
object HAAndUpdateStateByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("StreamingHAOfKafkaDirect")
      .setMaster("local[*]")
    val sc = SparkContext.getOrCreate(conf)
    val checkpointDir = "hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/012"

    // ========================================
    /**
      * 创建一个StreamingContext对象
      *
      * @return
      */
    def createingStreamingContextFunc(): StreamingContext = {
      val kafkaParams = Map("metadata.broker.list" -> "linux-hadoop01.ibeifeng.com:9092,linux-hadoop01.ibeifeng.com:9093,linux-hadoop01.ibeifeng.com:9094", "auto.offset.reset" -> "smallest")
      val topics = Set("beifeng")
      val ssc = new StreamingContext(sc, Seconds(1))
      val dstream = KafkaUtils.createDirectStream[String, String, kafka.serializer.StringDecoder, kafka.serializer.StringDecoder](ssc, kafkaParams, topics)

      val resultDStream = dstream
        .filter(_._2.nonEmpty)
        .mapPartitions(iter => {
          iter.flatMap(_._2.split(" ").map((_, 1)))
        })
        .updateStateByKey(
          (values: Seq[Int], state: Option[Long]) => {
            // 获取当前key的累加值
            val currentSum = values.sum
            // 获取之前批次的累加值
            val preSum = state.getOrElse(0L)

            // 返回结果
            Some(currentSum + preSum)
          }
        )

      // 打印一下
      resultDStream.print()

      // ssc设置checkpoint
      ssc.checkpoint(checkpointDir)
      // 返回对象
      ssc
    }

    val ssc = StreamingContext.getOrCreate(
      checkpointPath = checkpointDir,
      creatingFunc = createingStreamingContextFunc
    )

    // 将ssc传给其它管理Streaming关闭的对象==>其它线程(当前线程做数据的接收处理工作)
    shutdownStreaming(ssc)

    // start ===spark.streaming.stopGracefullyOnShutdown为true的时候，当jvm退出的时候，自动关闭
    ssc.start()
    ssc.awaitTermination() // 阻塞操作
  }

  /**
    * 进行shutdown操作
    *
    * @param ssc
    */
  def shutdownStreaming(ssc: StreamingContext): Unit = {
    new Thread(new Runnable {
      override def run(): Unit = {
        println("你好......")
        // 当某个条件被触发的时候，streamingcontext关闭
        // 这里假设停留10秒中
        Thread.sleep(10000)
        // 关闭操作
        ssc.stop(true, true)
        println("你好")
      }
    }).start()
  }

}

2.效果