sparkstreaming的reduceByKey(）算子 和updateStateByKey() 的差別

reduceByKey(）： 隻計算目前Duration時間内的聚合

updateStateByKey() ： 計算從streamingContext 啟動開始到目前批次的聚合，目前批次之前的資料儲存在記憶體+checkPoint 設定目錄中，不設定checkPoint 會報錯

如果Duration > 10s ， 每隔Duration時間做一次checkPoint

如果Duration <　10s ， 每隔１０ｓ時間做一次checkPoint，防止頻繁通路checkPoint 目錄

以下是reduceByKey updateStateByKey 使用代碼

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

    //receiver模式下接受資料，local的模拟線程必須大于等于2，一個線程用來receiver用來接受資料，另一個線程用來執行job。
    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreamingTest")

    //設定日志級别為ERROR 
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")

    //在建立streaminContext的時候 設定batch Interval
    val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))
    
    //建立DStream
    val dstream1: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop-101", 999)

    //執行DStream的transformation算子
    val dstream2: DStream[String] = dstream1.flatMap(x => {
      x.split(" ")
    })

    val dstream3: DStream[(String, Int)] = dstream2.map(x => {
      (x, 1)
    })

    val reducedStream: DStream[(String, Int)] = dstream3.reduceByKey(_ + _)

    //所有的代碼邏輯完成後要有一個output operation類算子觸發執行
    reducedStream.print()

    //Streaming架構啟動後不能再次添加業務邏輯。
    ssc.start()

    //等待reciverTask結束
    ssc.awaitTermination()


  }

}
           

object SparkStreamingTest2 extends App {

  val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreamingTest")

  //設定日志級别為ERROR
  val sc = new SparkContext(conf)
  sc.setLogLevel("ERROR")

  //在建立streaminContext的時候 設定batch Interval
  val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))

  ssc.checkpoint("./")

  //建立DStream
  val dstream1: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop-101", 999)

  //執行DStream的transformation算子
  val dstream2: DStream[String] = dstream1.flatMap(x => {
    x.split(" ")
  })

  val dstream3: DStream[(String, Int)] = dstream2.map(x => {
    (x, 1)
  })

  //解析updateStateByKey
  //updateStateByKey需要傳入一個函數(updateFunc: (Seq[V] Option[S])=>Option[S])
  //針對某個KEY
  //seq[v]: 是目前批次的某個key的資料:(1,1,1)
  //參數Option[S]): 是之前批次的這個key的累加資料：8
  //傳回值Option 是吧目前批次和累加批次聚合的結果
   val total: DStream[(String, Int)] = dstream3.updateStateByKey((currValues: Seq[Int], prevValueState: Option[Int]) => {
    val currentSum = currValues.sum
    //a a a
    val previousCount = prevValueState.getOrElse(0) //8
    Some(currentSum + previousCount) //3 + 8
  })

  total.print()
  ssc.start()
  ssc.awaitTermination()

}