SparkStreaming轉化操做

DStream的轉化操做分爲無狀態 和有狀態 兩種緩存

在無狀態轉化操做中，每一個批次的處理不依賴於以前批次的數據。
有狀態轉化操做須要使用以前批次的數據或者中間結果來計算當前批次的數據，有狀態轉化操做包括基於滑動窗口的轉化操做和追蹤狀態變化的轉換操做。

無狀態轉化

無狀態轉化操做的實質就說把簡單的RDD轉化操做應用到每一個批次上，也就是轉化DStream的每個RDDsocket

Transform算子

Transform 容許 DStream 上執行任意的 RDD-to-RDD 函數。即便這些函數並無在 DStream 的 API 中暴露出來，經過該函數能夠方便的擴展 Spark API。該函數每一批次調度一次。其實也就是對 DStream 中的 RDD 應用轉換。函數

def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("transform").setMaster("local[*]")
    val sc: StreamingContext = new StreamingContext(conf, Seconds(3))
    val lines = sc.socketTextStream("localhost", 9999)

    // transform方法能夠將底層RDD獲取到後進行操做
    // 1. DStream功能不完善
    // 2. 須要代碼週期性的執行

    // Code : Driver端
    val newDS: DStream[String] = lines.transform(
      rdd => {
        // Code : Driver端，（週期性執行）
        rdd.map(
          str => {
            // Code : Executor端
            str
          }
        )
      }
    )
    // Code : Driver端
    val newDS1: DStream[String] = lines.map(
      data => {
        // Code : Executor端
        data
      }
    )
    sc.start()
    sc.awaitTermination()
  }

join算子

兩個流之間的 join 須要兩個流的批次大小一致，這樣才能作到同時觸發計算。計算過程就是對當前批次的兩個流中各自的 RDD 進行 join，與兩個 RDD 的 join 效果相同。性能

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

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    val data9999 = ssc.socketTextStream("localhost", 9999)
    val data8888 = ssc.socketTextStream("localhost", 8888)

    val map9999: DStream[(String, Int)] = data9999.map((_,9))
    val map8888: DStream[(String, Int)] = data8888.map((_,8))

    // 所謂的DStream的Join操做，其實就是兩個RDD的join
    val joinDS: DStream[(String, (Int, Int))] = map9999.join(map8888)

    joinDS.print()

    ssc.start()
    ssc.awaitTermination()
}

有狀態轉化

有狀態轉化操做是跨時間區間跟蹤數據的操做，也就是說，一些先前批次的數據也被用來在新的批次中用於計算結果。有狀態轉換的主要的兩種類型:spa

滑動窗口：以一個時間階段爲滑動窗口進行操做
updateStateByKey()：經過key值來跟蹤數據的狀態變化

有狀態轉化操做須要在StreamingContext中打開檢查點機制來提升容錯code

updateStateByKey

def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("updateStateByKey")
    val sc: StreamingContext = new StreamingContext(conf, Seconds(4))
    sc.checkpoint("cp")
    val ds: ReceiverInputDStream[String] = sc.socketTextStream("localhost", 9999)

    val value: DStream[(String, Int)] = ds.map(((_: String), 1))


    // updateStateByKey：根據key對數據的狀態進行更新
    // 傳遞的參數中含有兩個值
    // 第一個值表示相同的key的value數據的集合
    // 第二個值表示緩存區key對應的計算值
    val state: DStream[(String, Int)] = value.updateStateByKey((seq: Seq[Int], option: Option[Int]) => {
      val newCount: Int = option.getOrElse(0) + seq.sum
      Option(newCount)
    })

    state.print()

    sc.start()
    sc.awaitTermination()

  }

窗口

全部基於窗口的函數都須要兩個參數，分別對應窗口時長和滑動步長,而且二者都必須是SparkStreaming的批次間隔的整數倍。
窗口時長控制的是每次用來計算的批次的個數
滑動步長用於控制對新的DStream進行計算的間隔orm

window操做

基於window進行窗口內元素計數操做blog

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

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    val lines = ssc.socketTextStream("localhost", 9999)
    val wordToOne = lines.map((_,1))

    val windowDS: DStream[(String, Int)] = wordToOne.window(Seconds(6), Seconds(6))

    val wordToCount = windowDS.reduceByKey(_+_)

    wordToCount.print()

    ssc.start()
    ssc.awaitTermination()
  }

reduce操做

有逆操做規約是一種更高效的規約操做，經過只考慮新進入窗口的元素和離開窗口的元素，讓spark增量計算歸約的結果，其在代碼上的體現就是reduceFunc 和 invReduceFuncget

普通歸約操做it

def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.checkpoint("cp")

    val lines = ssc.socketTextStream("localhost", 9999)

    lines.reduceByWindow(
      (x: String, y: String) => {
        x + "-" + y
      },
      Seconds(9), Seconds(3)
    ).print()

    ssc.start()
    ssc.awaitTermination()
  }

有逆歸約操做

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

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.checkpoint("cp")

    val lines = ssc.socketTextStream("localhost", 9999)
    val wordToOne = lines.map((_,1))

    /**
     * 基於窗口進行有逆歸約：經過控制窗口流出和進入的元素來提升性能
     */
    val windowDS: DStream[(String, Int)] =
    wordToOne.reduceByKeyAndWindow(
      (x:Int, y:Int) => { x + y},
      (x:Int, y:Int) => {x - y},
      Seconds(9), Seconds(3))

    windowDS.print()

    ssc.start()
    ssc.awaitTermination()
  }

count操做

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

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.checkpoint("cp")

    val lines = ssc.socketTextStream("localhost", 9999)

    /**
     * 統計窗口中輸入數據的個數
     * 好比 3s內輸入了10條數據，則打印10
     */
    val countByWindow: DStream[Long] = lines.countByWindow(
      Seconds(9), Seconds(3)
    )
    countByWindow.print()

    /**
     * 統計窗口中每一個值的個數
     * 好比 3s內輸入了1個3 2個4 3個5，則打印(3,1)(2,4)(3,5)
     */
    val countByValueAndWindow: DStream[(String, Long)] = lines.countByValueAndWindow(
      Seconds(9), Seconds(3)
    )
    countByValueAndWindow.print()

    ssc.start()
    ssc.awaitTermination()
  }