Spark Streaming基本概念

一、關聯

與Spark類似，Spark Streaming也可以利用maven倉庫。編寫你自己的Spark Streaming程式，你需要引入下面的依賴到你的SBT或者Maven項目中
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.10</artifactId>
        <version>1.2</version>
    </dependency>

    為了從Kafka, Flume和Kinesis這些不在Spark核心API中提供的源擷取資料，我們需要添加相關的子產品spark-streaming-xyz_2.10 到依賴中
    以下為一些常用元件
        kafka:spark-streaming-kafka_2.10
        flume:spark-streaming-flume_2.10
        Kinesis:spark-streaming-kinesis-asl_2.10
        Twitter:spark-streaming-twitter_2.10
        ZeroMQ:spark-streaming-zeromq_2.10
        MQTT:spark-streaming-mqtt_2.10
           

二、初始化StreamingContext

為了初始化Spark Streaming程式，一個StreamingContext對象必需被建立，它是Spark Streaming所有流操作的主要入口。一個StreamingContext對象可以用SparkConf對象建立。
    import org.apache.spark._
    import org.apache.spark.streaming._
    val conf = new SparkConf().setAppName(appName).setMaster(master)
    val ssc = new StreamingContext(conf, Seconds(1))

    appName 表示你的應用程式顯示在叢集UI上的名字， master 是一個Spark、Mesos、YARN叢集URL或者一個特殊字元串“local[*]”，它表示程式用本地模式運作。當程式運作在叢集中時，你并不希望在程式中寫死 master ，而是希望用 spark-submit啟動應用程式，并從 spark-submit中得到 master 的值。對于本地測試或者單元測試，你可以傳遞“local”字元串在同一個程序内運作Spark Streaming。需要注意的是，它在内部建立了一個SparkContext對象，你可以通過 ssc.sparkContext 通路這個SparkContext對象。

    當一個上下文（context）定義之後，你必須按照以下幾步進行操作：
        1.定義輸入源
        2.準備好流計算指令
        3.利用 streamingContext.start() 方法接收和處理資料
        4.處理過程将一直持續，直到 streamingContext.stop() 方法被調用

    需要注意的地方：
        1.一旦一個context已經啟動，就不能有新的流算子建立或者是添加到context中。
        2.一旦一個context已經停止，它就不能再重新啟動
        3.在JVM中，同一時間隻能有一個StreamingContext處于活躍狀态
        4.在StreamingContext上調用 stop() 方法，也會關閉SparkContext對象。如果隻想僅關閉
        5.StreamingContext對象，設定 stop() 的可選參數為false
        6.一個SparkContext對象可以重複利用去建立多個StreamingContext對象，前提條件是前面的
        7.StreamingContext在後面StreamingContext建立之前關閉（不關閉SparkContext）。
           

三、離散流（DStreams）

離散流或者DStreams是Spark Streaming提供的基本的抽象，它代表一個連續的資料流。它要麼是從源中擷取的輸入流，要麼是輸入流通過轉換算子生成的處理後的資料流。在内部，DStreams由一系列連續的RDD組成。DStreams中的每個RDD都包含确定時間間隔内的資料。
    任何對DStreams的操作都轉換成了對DStreams隐含的RDD的操作。
           

四、輸入DStreams和receivers

輸入DStreams表示從資料源擷取輸入資料流的DStreams。在SparkStreaming快速例子中， lines 表示輸入DStream，它代表從netcat伺服器擷取的資料流。每一個輸入流DStream和一個 Receiver 對象相關聯，這個 Receiver從源中擷取資料，并将資料存入記憶體中用于處理。

輸入DStreams表示從資料源擷取的原始資料流。Spark Streaming擁有兩類資料源:
    基本源（Basic sources）：這些源在StreamingContext API中直接可用。例如檔案系統、套接字連接配接、Akka的actor等。
    進階源（Advanced sources）：這些源包括Kafka,Flume,Kinesis,Twitter等等。它們需要通過額外的類來使用。

需要注意的是，如果你想在一個流應用中并行地建立多個輸入DStream來接收多個資料流，你能夠建立多個輸入流。它将建立多個Receiver同時接收多個資料流。但是， receiver作為一個長期運作的任務運作在Spark worker或executor中。是以，它占有一個核，這個核是配置設定給Spark Streaming應用程式的所有核中的一個（it occupies one of the cores allocated to the SparkStreaming application）。是以，為Spark Streaming應用程式配置設定足夠的核（如果是本地運作，那麼是線程）用以處理接收的資料并且運作 receiver 是非常重要的。

幾點需要注意的地方：
    如果配置設定給應用程式的核的數量少于或者等于輸入DStreams或者receivers的數量，系統隻能夠接收資料而不能處理它們。
    當運作在本地，如果你的master URL被設定成了“local”，這樣就隻有一個核運作任務。這對程式來說是不足的，因為作為 receiver 的輸入DStream将會占用這個核，這樣就沒有剩餘的核來處理資料了。

基本源:
    檔案流（File Streams）：從任何與HDFS API相容的檔案系統中讀取資料，一個DStream可以通過如下方式建立。
        需要注意的地方：
            1.所有檔案必須具有相同的資料格式
            2.所有檔案必須在`dataDirectory`目錄下建立，檔案是自動的移動和重命名到資料目錄下
            3.一旦移動，檔案必須被修改。是以如果檔案被持續的附加資料，新的資料不會被讀取。

    基于自定義actor的流:
        DStream可以調streamingContext.actorStream(actorProps, actor-name) 方法從Akka actors擷取的資料流來建立。

    RDD隊列作為資料流：
        為了用測試資料測試Spark Streaming應用程式，人們也可以調用streamingContext.queueStream(queueOfRDDs) 方法基于RDD隊列建立DStreams。每個push到隊列的RDD都被當做DStream的批資料，像流一樣處理。

進階源:
    這類源需要非Spark庫接口，并且它們中的部分還需要複雜的依賴（例如kafka和flume）。

自定義源:
    在Spark 1.2中，這些源不被Python API支援。輸入DStream也可以通過自定義源建立，你需要做的是實作使用者自定義的 receiver ，這個 receiver 可以從自定義源接收資料以及将資料推到Spark中。
    Receiver可靠性
基于可靠性有兩類資料源。源(如kafka、flume)允許。如果從這些可靠的源擷取資料的系統能夠正确的應答所接收的資料，它就能夠確定在任何情況下不丢失資料。這樣，就有兩種類型的receiver：
        Reliable Receiver：一個可靠的receiver正确的應答一個可靠的源，資料已經收到并且被正确地複制到了Spark中。
        Unreliable Receiver ：這些receivers不支援應答。即使對于一個可靠的源，開發者可能實作一個非可靠的receiver，這個receiver不會正确應答。
           

五、DStream中的轉換（transformation）

和RDD類似，transformation允許從輸入DStream來的資料被修改。DStreams支援很多在RDD中可用的transformation算子。一些常用的算子如下所示：
    map(func): 利用函數 func 處理原DStream的每個元素，傳回一個新的DStream


    filter(func): 傳回一個新的DStream，它僅僅包含源DStream中滿足函數func的項

    repartition(numPartitions):通過建立更多或者更少的partition改變這個DStream的并行級别(level of parallelism)

    union(otherStream): 傳回一個新的DStream,它包含源DStream和otherStream的聯合元素

    count():通過計算源DStream中每個RDD的元素數量，傳回一個包含單元素(single-element)RDDs的新DStream

    reduce(func):利用函數func聚集源DStream中每個RDD的元素，傳回一個包含單元素(single-element)RDDs的新DStream。函數應該是相關聯的，以使計算可以并行化

    countByValue():這個算子應用于元素類型為K的DStream上，傳回一個（K,long）對的新DStream，每個鍵的值是在原DStream的每個RDD中的頻率。


    join(otherStream,[numTasks]):當應用于兩個DStream（一個包含（K,V）對,一個包含(K,W)對），傳回一個包含(K, (V, W))對的新DStream


    transform(func):通過對源DStream的每個RDD應用RDD-to-RDD函數，建立一個新的DStream。這個可以在DStream中的任何RDD操作中使用

    updateStateByKey(func):利用給定的函數更新DStream的狀态，傳回一個新"state"的DStream。

    重點介紹下面兩個算子：
        updateStateByKey：操作允許不斷用新資訊更新它的同時保持任意狀态。你需要通過兩步來使用它
            1.定義狀态-狀态可以是任何的資料類型
            2.定義狀态更新函數-怎樣利用更新前的狀态和從輸入流裡面擷取的新值更新狀态

        示例如下：
        val sparkConf = new SparkConf().setAppName("StatefulNetworkWordCount") 
        // Create the context with a 1 second batch size 
        val ssc = new StreamingContext(sparkConf, Seconds(1)) 
        ssc.checkpoint(".") 


        // Initial state RDD for mapWithState operation 
        val initialRDD = ssc.sparkContext.parallelize(List(("hello", 1), ("world", 1))) 


        // Create a ReceiverInputDStream on target ip:port and count the 
        // words in input stream of \n delimited test (eg. generated by 'nc') 
        val lines = ssc.socketTextStream(args(0), args(1).toInt) 
        val words = lines.flatMap(_.split(" ")) 
        val wordDstream = words.map(x => (x, 1)) 


        // Update the cumulative count using mapWithState 
        // This will give a DStream made of state (which is the cumulative count of the words) 
        val mappingFunc = (word: String, one: Option[Int], state: State[Int]) => { 
        val sum = one.getOrElse(0) + state.getOption.getOrElse(0) 
        val output = (word, sum) 
        state.update(sum) 
        output 
        } 


        val stateDstream = wordDstream.mapWithState( 
        StateSpec.function(mappingFunc).initialState(initialRDD)) 
        stateDstream.print() 
        ssc.start() 
        ssc.awaitTermination() 
        } 
    Transform操作：
        transform 操作（以及它的變化形式如 transformWith ）允許在DStream運作任何RDD-to-RDD函數。它能夠被用來應用任何沒在DStream API中提供的RDD操作（It can be used to apply any RDDoperation that is not exposed in the DStream API）。例如，連接配接資料流中的每個批（batch）和另外一個資料集的功能并沒有在DStream API中提供，然而你可以簡單的利用 transform 方法做到。如果你想通過連接配接帶有預先計算的垃圾郵件資訊的輸入資料流來清理實時資料，然後過了它們，你可以按如下方法來做：
    val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // RDD containing spam informa
    tion
    val cleanedDStream = wordCounts.transform(rdd => {
    rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
    ...
    })
           

六、DStreams上的輸出操作

輸出操作允許DStream的操作推到如資料庫、檔案系統等外部系統中。因為輸出操作實際上是允許外部系統消費轉換後的資料，它們觸發的實際操作是DStream轉換。目前，定義了下面幾種輸出操作：
    print():
    在DStream的每個批資料中列印前10條元素，這個操作在開發和調試中都非常有用。在Python API中調用 pprint() 。

    saveAsObjectFiles(prefix,[suffix]):

    saveAsTextFiles(prefix,[suffix]):
    儲存DStream的内容為一個文本檔案。每一個批間隔的檔案的檔案名基于 prefix 和 suffix 生成。"prefix-TIME_IN_MS[.suffix]"

    saveAsHadoopFiles(prefix,[suffix]):
    儲存DStream的内容為一個hadoop檔案。每一個批間隔的檔案的檔案名基于 prefix 和 suffix 生成。"prefix-TIME_IN_MS[.suffix]"，在Python API中不可用。
    foreachRDD(func):
    在從流中生成的每個RDD上應用函數 func 的最通用的輸出操作。這個函數應該推送每個RDD的資料到外部系統，例如儲存RDD到檔案或者通過網絡寫到資料庫中。需要注意的是， func 函數在驅動程式中執行，并且通常都有RDD action在裡面推動RDD流的計算。

利用foreachRDD的設計模式
    dstream.foreachRDD是一個強大的原語，發送資料到外部系統中。然而，明白怎樣正确地、有效地用這個原語是非常重要的。下面幾點介紹了如何避免一般錯誤。經常寫資料到外部系統需要建一個連接配接對象（例如到遠端伺服器的TCP連接配接），用它發送資料到遠端系統。為了達到這個目的，開發人員可能不經意的在Spark驅動中建立一個連接配接對象，但是在Sparkworker中嘗試調用這個連接配接對象儲存記錄到RDD中，如下：
    dstream.foreachRDD(rdd => {
    val connection = createNewConnection() // executed at the driver
    rdd.foreach(record => {
    connection.send(record) // executed at the worker
    })
    })
    這是不正确的，因為這需要先序列化連接配接對象，然後将它從driver發送到worker中。這樣的連接配接對象在機器之間不能傳送。它可能表現為序列化錯誤（連接配接對象不可序列化）或者初始化錯誤（連接配接對象應該在worker中初始化）等等。正确的解決辦法是在worker中建立連接配接對象。然而，這會造成另外一個常見的錯誤-為每一個記錄建立了一個連接配接對象。例如：

    dstream.foreachRDD(rdd => {
    rdd.foreach(record => {
    val connection = createNewConnection()
    connection.send(record)
    connection.close()
    })
    })
    通常，建立一個連接配接對象有資源和時間的開支。是以，為每個記錄建立和銷毀連接配接對象會導緻非常高的開支，明顯的減少系統的整體吞吐量。一個更好的解決辦法是利用 rdd.foreachPartition 方法。為RDD的partition建立一個連接配接對象，用這個兩件對象發送partition中的所有記錄。
    dstream.foreachRDD(rdd => {
    rdd.foreachPartition(partitionOfRecords => {
    val connection = createNewConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    connection.close()
    })
    })
    這就将連接配接對象的建立開銷分攤到了partition的所有記錄上了。最後，可以通過在多個RDD或者批資料間重用連接配接對象做更進一步的優化。開發者可以保有一個靜态的連接配接對象池，重複使用池中的對象将多批次的RDD推送到外部系統，以進一步節省開支。
    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
    })
    })
    需要注意的是，池中的連接配接對象應該根據需要延遲建立，并且在空閑一段時間後自動逾時。這樣就擷取了最有效的方式發生資料到外部系統。
    其它需要注意的地方：
    輸出操作通過懶執行的方式操作DStreams，正如RDD action通過懶執行的方式操作RDD。具體地看，RDD actions和DStreams輸出操作接收資料的處理。是以，如果你的應用程式沒有任何輸出操作或者用于輸出操作 dstream.foreachRDD() ，但是沒有任何RDD action操作在dstream.foreachRDD() 裡面，那麼什麼也不會執行。系統僅僅會接收輸入，然後丢棄它們。
    預設情況下，DStreams輸出操作是分時執行的，它們按照應用程式的定義順序按序執行。