spark系列-8、Spark Streaming

• 2020 年 5 月 5 日
• 筆記
• 大數據之spark

參考鏈接：//spark.apache.org/docs/latest/streaming-programming-guide.html

一、Spark Streaming 介紹

Spark Streaming是核心Spark API的擴展，可實現實時數據流的可伸縮，高吞吐量，容錯流處理。數據可以從Kafka、ZeroMQ等消息隊列以及TCP sockets或者目錄文件從數據源獲取數據，並且可以使用map，reduce，join和window等高級函數進行複雜演算法的處理。最後，可以將處理後的數據推送到文件系統，資料庫和實時儀錶板。

• 在內部，它的工作方式為：Spark Streaming接收實時輸入數據流，並將數據分成批次，然後由Spark引擎進行處理，以生成批次的最終結果流。
• 
• 對應的批數據，在Spark內核對應一個RDD實例，因此，對應流數據的DStream可以看成是一組RDDs，即RDD的一個序列。通俗點理解的話，在流數據分成一批一批後，通過一個先進先出的隊列，然後 Spark Engine從該隊列中依次取出一個個批數據，把批數據封裝成一個RDD，然後進行處理，這是一個典型的生產者消費者模型，對應的就有生產者消費者模型的問題，即如何協調生產速率和消費速率。 

Spark Streaming Wordcount：

• import org.apache.spark.SparkConf
  import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
  
  /**
    * @author xiandongxie
    */
  object SparkStreamingWordCount {
    def main(args: Array[String]): Unit = {
      val conf: SparkConf = new SparkConf()
        .setMaster("local[2]")
        .setAppName("SparkStreamingWordCount")
      // 設置批次時間5S
      val duration: Duration = Durations.seconds(5)
      val context: StreamingContext = new StreamingContext(conf,duration)
      // 指定socket數據源
      val sourceDStream: ReceiverInputDStream[String] = context.socketTextStream("localhost", 6666)
      // 計算WordCount
      val resultDStream: DStream[(String, Int)] = sourceDStream.flatMap(f => f.split("\t"))
        .map((_, 1))
        .reduceByKey(_ + _)
  
      resultDStream.print()
  
      context.start()             // Start the computation
      context.awaitTermination()  // Wait for the computation to terminate
    }
  }

二、Spark Streaming 對比 Storm

• 處理模型以及延遲
  • 雖然兩框架都提供了可擴展性(scalability)和可容錯性(fault tolerance)，但是它們的處理模型從根本上說是不一樣的。Storm可以實現亞秒級時延的處理，而每次只處理一條event，而Spark Streaming可以在一個短暫的時間窗口裡面處理多條(batches)Event。所以說Storm可以實現亞秒級時延的處理，而Spark Streaming則有一定的時延。
• 容錯和數據保證
  • 然而兩者的都有容錯時候的數據保證，Spark Streaming的容錯（通過血緣關係）為有狀態的計算提供了更好的支援。在Storm中，每條記錄在系統的移動過程中都需要被標記跟蹤，所以Storm只能保證每條記錄最少被處理一次，但是允許從錯誤狀態恢復時被處理多次。這就意味著可變更的狀態可能被更新兩次從而導致結果不正確。
  • Spark Streaming的容錯：通過血緣關係，是粗粒度的，保證每個批處理記錄僅僅被處理一次，即使是node節點掛掉
  • Storm：細粒度的容錯，每條記錄在系統的移動過程中都需要被標記跟蹤，缺點：允許從錯誤狀態恢復時被處理多次。這就意味著可變更的狀態可能被更新兩次從而導致結果不正確。
• 批處理框架集成
  • Spark Streaming的一個很棒的特性就是它是在Spark框架上運行的。這樣你就可以使用spark的批處理程式碼一樣來寫Spark Streaming程式，或者是在Spark中交互查詢比如spark-sql。這就減少了單獨編寫流處理程式和歷史數據處理程式。
• 生產支援
  • 兩者都可以在各自的集群框架中運行，但是Storm可以在Mesos上運行, 而Spark Streaming可以在YARN和Mesos上運行。

 Spark Streaming優缺點：

• 優點：
  • 吞吐量大、速度快。
  • 容錯：SparkStreaming在沒有額外程式碼和配置的情況下可以恢復丟失的工作。checkpoint。
  • 社區活躍度高。生態圈強大。因為後台是Spark
  • 數據源廣泛。
• 缺點：
  • 延遲。500毫秒已經被廣泛認為是最小批次大小。所以實際場景中應注意該問題，就像標題分類場景，設定的0.5s一批次，加上處理時間，分類介面會佔用1s的響應時間。實時要求高的可選擇使用其他框架。

三、架構與抽象

Spark Streaming使用「微批次」的架構，把流試計算當成一系列連接的小規模批處理來對待，Spark Streaming從各種輸入源中讀取數據，並把數據分成小組的批次，新的批次按均勻的時間間隔創建出來，在每個時間區間開始的時候，一個新的批次就創建出來，在該區間內收到的數據都會被添加到這個批次中，在時間區間結束時，批次停止增長。時間區間的大小是由批次間隔這個參數決定的，批次間隔一般設在500毫秒到幾秒之間，由應用開發者配置，每個輸出批次都會形成一個RDD，以Spark作業的方式處理並生成其他的RDD。並能將處理結果按批次的方式傳給外部系統。

接受器（receive）會佔用一個executor的一個cpu，所以在local[n]模式下，n > 要運行的接收器數

四、DStream 操作

• DStream 上的原語與 RDD 的類似，分為 Transformations（轉換，惰性的）和 Output Operations（輸出）兩種，此外轉換操作中還有一些比較特殊的原語，如：
  • updateStateByKey()、transform() 以及各種 Window 相關的原語。
• UpdateStateByKey 返回一個新的「狀態」 DStream，在該DStream中，通過在鍵的先前狀態和鍵的新值上應用給定函數來更新每個鍵的狀態。這可用於維護每個鍵的任意狀態數據。
  • 如輸入：hello world，結果則為：(hello,1)(world,1)，然後輸入 hello spark，結果則為 (hello,2)(spark,1)。會保留上一次數據處理的結果。

  • import org.apache.spark.SparkConf
    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{Durations, StreamingContext}
    
    /**
      * @author xiandongxie
      */
    object SparkStreamingSocketPortUpdateState {
      def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
          .setAppName("sparkstreamingsocketportupdatestate")
          .setMaster("local[2]")
        val streamingContext = new StreamingContext(conf, Durations.seconds(5))
    
        // 設置保存地址
        streamingContext.checkpoint("/tmp/spark/sparkstreamingsocketportupdatestat")
    
        val sourceDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("localhost", 6666)
    
        val reduceDStream: DStream[(String, Int)] = sourceDStream.flatMap(_.split(" "))
          .map((_, 1))
          .reduceByKey(_ + _)
    
        val updateStateByKey: DStream[(String, Int)] = reduceDStream.updateStateByKey((newValues: Seq[Int], runningCount: Option[Int]) => {
          var total: Int = 0
          for (i <- newValues) {
            total += i
          }
          println(newValues  + "\t" + (if (runningCount.isDefined) runningCount.get else 0))
          val last: Int = if (runningCount.isDefined) runningCount.get else 0
          val now: Int = total + last
          Some(now)
        })
    
        updateStateByKey.foreachRDD((r, t) => {
          println(s"count time:${t},${r.collect().toList}")
        })
    
    
        streamingContext.start()
        streamingContext.awaitTermination()
    
      }
    }

• Transform() 原語允許 DStream 上執行任意的 RDD-to-RDD 函數 代替原有的DStream 轉換操作，必須返回一個RDD；通過該函數可以方便的擴展 Spark API。

普通的轉換操作如下表所示：

五、窗口轉換操作

• 在Spark Streaming中，數據處理是按批進行的，而數據採集是逐條進行的，因此在Spark Streaming中會先設置好批處理間隔（batch duration），當超過批處理間隔的時候就會把採集到的數據匯總起來成為一批數據交給系統去處理。
• 對於窗口操作而言，在其窗口內部會有N個批處理數據，批處理數據的大小由窗口間隔（window duration）決定，而窗口間隔指的就是窗口的持續時間，在窗口操作中，只有窗口的長度滿足了才會觸發批數據的處理。除了窗口的長度，窗口操作還有另一個重要的參數就是滑動間隔（slide duration），它指的是經過多長時間窗口滑動一次形成新的窗口，滑動窗口默認情況下和批次間隔的相同，而窗口間隔一般設置的要比它們兩個大。在這裡必須注意的一點是滑動間隔和窗口間隔的大小一定得設置為批處理間隔的整數倍
• 

val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

窗口的計算：

Spark Streaming 還提供了窗口的計算，它允許你通過滑動窗口對數據進行轉換，窗口轉換操作如下：

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }
}

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}

/**
  * @author xiandongxie
  */
object SparkStreamingSocketCheckPoint {
  def main(args: Array[String]): Unit = {
    val checkpointPath = "/tmp/spark/sparkStreamingSocketCheckPoint"

    val strc: StreamingContext = StreamingContext.getOrCreate(checkpointPath, () => {
      val conf: SparkConf = new SparkConf()
        .setAppName("SparkStreamingSocketCheckPoint")
        .setMaster("local[2]")
      val streamingContext = new StreamingContext(conf, Durations.seconds(5))
      streamingContext.checkpoint(checkpointPath)

      val sourceDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("localhost", 6666)

      val reduceDStream: DStream[(String, Int)] = sourceDStream.flatMap(_.split(" "))
        .map((_, 1))
        .reduceByKey(_ + _)

      val updateStateByKey: DStream[(String, Int)] = reduceDStream.updateStateByKey((newValues: Seq[Int], runningCount: Option[Int]) => {
        var total: Int = 0
        for (i <- newValues) {
          total += i
        }
        val last: Int = if (runningCount.isDefined) runningCount.get else 0
        val now: Int = total + last
        Some(now)
      })

      updateStateByKey.foreachRDD((r, t) => {
        println(s"count time:${t},${r.collect().toList}")
      })
      streamingContext
    })

    strc.start()
    strc.awaitTermination()
  }
}