spark-1.6.1 [原文地址]

Spark Streaming程式設計指南

概覽

Spark Streaming是對核心Spark API的一個擴充套件，它能夠實現對實時資料流的流式處理，並具有很好的可擴充套件性、高吞吐量和容錯性。Spark Streaming支援從多種資料來源提取資料，如：Kafka、Flume、Twitter、ZeroMQ、Kinesis以及TCP套接字，並且可以提供一些高階API來表達複雜的處理演算法，如：map、reduce、join和window等。最後，Spark Streaming支援將處理完的資料推送到檔案系統、資料庫或者實時儀表盤中展示。實際上，你完全可以將Spark的機器學習（

下圖展示了Spark Streaming的內部工作原理。Spark Streaming從實時資料流接入資料，再將其劃分為一個個小批量供後續Spark engine處理，所以實際上，Spark Streaming是按一個個小批量來處理資料流的。

Spark Streaming為這種持續的資料流提供了的一個高階抽象，即：discretized stream（離散資料流）或者叫DStream。DStream既可以從輸入資料來源建立得來，如：Kafka、Flume或者Kinesis，也可以從其他DStream經一些運算元操作得到。其實在內部，一個DStream就是包含了一系列

本文件將向你展示如何用DStream進行Spark Streaming程式設計。Spark Streaming支援Scala、Java和Python（始於Spark 1.2），本文件的示例包括這三種語言。

注意：對Python來說，有一部分API尚不支援，或者是和Scala、Java不同。本文件中會用高亮形式來註明這部分 Python API。

一個小栗子

在深入Spark Streaming程式設計細節之前，我們先來看看一個簡單的小栗子以便有個感性認識。假設我們在一個TCP埠上監聽一個數據伺服器的資料，並對收到的文字資料中的單詞計數。以下你所需的全部工作：

• Scala
• Java

首先，我們需要匯入Spark Streaming的相關class的一些包，以及一些支援StreamingContext隱式轉換的包（這些隱式轉換能給DStream之類的class增加一些有用的方法）。StreamingContext 是Spark Streaming的入口。我們將會建立一個本地 StreamingContext物件，包含兩個執行執行緒，並將批次間隔設為1秒。

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._ // 從Spark 1.3之後這行就可以不需要了

// 建立一個local StreamingContext，包含2個工作執行緒，並將批次間隔設為1秒
// master至少需要2個CPU核，以避免出現任務餓死的情況
val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1))

利用這個上下文物件（StreamingContext），我們可以建立一個DStream，該DStream代表從前面的TCP資料來源流入的資料流，同時TCP資料來源是由主機名（如：hostnam）和埠（如：9999）來描述的。

// 建立一個連線到hostname:port的DStream，如：localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)

這裡的 lines 就是從資料server接收到的資料流。其中每一條記錄都是一行文字。接下來，我們就需要把這些文字行按空格分割成單詞。

// 將每一行分割成多個單詞
val words = lines.flatMap(_.split(" "))

flatMap 是一種 “一到多”（one-to-many）的對映運算元，它可以將源DStream中每一條記錄對映成多條記錄，從而產生一個新的DStream物件。在本例中，lines中的每一行都會被flatMap對映為多個單詞，從而生成新的words DStream物件。然後，我們就能對這些單詞進行計數了。

import org.apache.spark.streaming.StreamingContext._ // Spark 1.3之後不再需要這行
// 對每一批次中的單詞進行計數
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

// 將該DStream產生的RDD的頭十個元素列印到控制檯上
wordCounts.print()

words這個DStream物件經過map運算元（一到一的對映）轉換為一個包含（word, 1）鍵值對的DStream物件pairs，再對pairs使用reduce運算元，得到每個批次中各個單詞的出現頻率。最後，wordCounts.print() 將會每秒（前面設定的批次間隔）列印一些單詞計數到控制檯上。

注意，執行以上程式碼後，Spark Streaming只是將計算邏輯設定好，此時並未真正的開始處理資料。要啟動之前的處理邏輯，我們還需要如下呼叫：

ssc.start()             // 啟動流式計算
ssc.awaitTermination()  // 等待直到計算終止

完整的程式碼可以在Spark Streaming的例子 NetworkWordCount 中找到。

如果你已經有一個Spark包（下載在這裡downloaded，自定義構建在這裡built），就可以執行按如下步驟執行這個例子。

首先，你需要執行netcat（Unix-like系統都會有這個小工具），將其作為data server

$ nc -lk 9999

然後，在另一個終端，按如下指令執行這個例子

• Scala
• Java

$ ./bin/run-example streaming.NetworkWordCount localhost 9999

好了，現在你嘗試可以在執行netcat的終端裡敲幾個單詞，你會發現這些單詞以及相應的計數會出現在啟動Spark Streaming例子的終端螢幕上。看上去應該和下面這個示意圖類似：

# TERMINAL 1:
# Running Netcat

$ nc -lk 9999

hello world



...

$ ./bin/run-example streaming.NetworkWordCount localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...

基本概念

下面，我們在之前的小栗子基礎上，繼續深入瞭解一下Spark Streaming的一些基本概念。

連結依賴項

和Spark類似，Spark Streaming也能在Maven庫中找到。如果你需要編寫Spark Streaming程式，你就需要將以下依賴加入到你的SBT或Maven工程依賴中。

• Maven
• SBT

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.10</artifactId>
    <version>1.6.1</version>
</dependency>

還有，對於從Kafka、Flume以及Kinesis這類資料來源提取資料的流式應用來說，還需要額外增加相應的依賴項，下表列出了各種資料來源對應的額外依賴項：

最新的依賴項資訊（包括原始碼和Maven工件）請參考Maven repository。

初始化StreamingContext

要初始化任何一個Spark Streaming程式，都需要在入口程式碼中建立一個StreamingContext物件。

• Scala
• Java

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 是你給該應用起的名字，這個名字會展示在Spark叢集的web UI上。而 master 是Spark, Mesos or YARN cluster URL，如果支援本地測試，你也可以用”local[*]”為其賦值。通常在實際工作中，你不應該將master引數硬編碼到程式碼裡，而是應用通過spark-submit的引數來傳遞master的值（launch the application with spark-submit ）。不過對本地測試來說，”local[*]”足夠了（該值傳給master後，Spark Streaming將在本地程序中，啟動n個執行緒執行，n與本地系統CPU core數相同）。注意，StreamingContext在內部會建立一個  SparkContext 物件（SparkContext是所有Spark應用的入口，在StreamingContext物件中可以這樣訪問：ssc.sparkContext）。

StreamingContext還有另一個構造引數，即：批次間隔，這個值的大小需要根據應用的具體需求和可用的叢集資源來確定。詳見Spark效能調優（ Performance Tuning）。

StreamingContext物件也可以通過已有的SparkContext物件來建立，示例如下：

import org.apache.spark.streaming._

val sc = ...                // 已有的SparkContext
val ssc = new StreamingContext(sc, Seconds(1))

context物件建立後，你還需要如下步驟：

1. 建立DStream物件，並定義好輸入資料來源。
2. 基於資料來源DStream定義好計算邏輯和輸出。
3. 呼叫streamingContext.start() 啟動接收並處理資料。
4. 呼叫streamingContext.awaitTermination() 等待流式處理結束（不管是手動結束，還是發生異常錯誤）
5. 你可以主動呼叫 streamingContext.stop() 來手動停止處理流程。

需要關注的重點:

• 一旦streamingContext啟動，就不能再對其計算邏輯進行新增或修改。
• 一旦streamingContext被stop掉，就不能restart。
• 單個JVM虛機同一時間只能包含一個active的StreamingContext。
• StreamingContext.stop() 也會把關聯的SparkContext物件stop掉，如果不想把SparkContext物件也stop掉，可以將StreamingContext.stop的可選引數 stopSparkContext 設為false。
• 一個SparkContext物件可以和多個StreamingContext物件關聯，只要先對前一個StreamingContext.stop(sparkContext=false)，然後再建立新的StreamingContext物件即可。

離散資料流 (DStreams)

離散資料流（DStream）是Spark Streaming最基本的抽象。它代表了一種連續的資料流，要麼從某種資料來源提取資料，要麼從其他資料流對映轉換而來。DStream內部是由一系列連續的RDD組成的，每個RDD都是不可變、分散式的資料集（詳見Spark程式設計指南 – Spark Programming Guide）。每個RDD都包含了特定時間間隔內的一批資料，如下圖所示：

任何作用於DStream的運算元，其實都會被轉化為對其內部RDD的操作。例如，在前面的例子中，我們將 lines 這個DStream轉成words DStream物件，其實作用於lines上的flatMap運算元，會施加於lines中的每個RDD上，並生成新的對應的RDD，而這些新生成的RDD物件就組成了words這個DStream物件。其過程如下圖所示：

底層的RDD轉換仍然是由Spark引擎來計算。DStream的運算元將這些細節隱藏了起來，併為開發者提供了更為方便的高階API。後續會詳細討論這些高階運算元。

輸入DStream和接收器

輸入DStream代表從某種流式資料來源流入的資料流。在之前的例子裡，lines 物件就是輸入DStream，它代表從netcat server收到的資料流。每個輸入DStream（除檔案資料流外）都和一個接收器（Receiver – Scala doc, Java doc）相關聯，而接收器則是專門從資料來源拉取資料到記憶體中的物件。

Spark Streaming主要提供兩種內建的流式資料來源：

• 基礎資料來源（Basic sources）: 在StreamingContext API 中可直接使用的源，如：檔案系統，套接字連線或者Akka actor。
• 高階資料來源（Advanced sources）: 需要依賴額外工具類的源，如：Kafka、Flume、Kinesis、Twitter等資料來源。這些資料來源都需要增加額外的依賴，詳見依賴連結（linking）這一節。

本節中，我們將會從每種資料來源中挑幾個繼續深入討論。

注意，如果你需要同時從多個數據源拉取資料，那麼你就需要建立多個DStream物件（詳見後續的效能調優這一小節）。多個DStream物件其實也就同時建立了多個數據流接收器。但是請注意，Spark的worker/executor 都是長期執行的，因此它們都會各自佔用一個分配給Spark Streaming應用的CPU。所以，在執行Spark Streaming應用的時候，需要注意分配足夠的CPU core（本地執行時，需要足夠的執行緒）來處理接收到的資料，同時還要足夠的CPU core來執行這些接收器。

要點

• 如果本地執行Spark Streaming應用，記得不能將master設為”local” 或 “local[1]”。這兩個值都只會在本地啟動一個執行緒。而如果此時你使用一個包含接收器（如：套接字、Kafka、Flume等）的輸入DStream，那麼這一個執行緒只能用於執行這個接收器，而處理資料的邏輯就沒有執行緒來執行了。因此，本地執行時，一定要將master設為”local[n]”，其中 n > 接收器的個數（有關master的詳情請參考Spark Properties）。
• 將Spark Streaming應用置於叢集中執行時，同樣，分配給該應用的CPU core數必須大於接收器的總數。否則，該應用就只會接收資料，而不會處理資料。

基礎資料來源

前面的小栗子中，我們已經看到，使用ssc.socketTextStream(…) 可以從一個TCP連線中接收文字資料。而除了TCP套接字外，StreamingContext API 還支援從檔案或者Akka actor中拉取資料。

• 檔案資料流（File Streams）: 可以從任何相容HDFS API（包括：HDFS、S3、NFS等）的檔案系統，建立方式如下：
  • Scala
  • Java

    streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)

  Spark Streaming將監視該dataDirectory目錄，並處理該目錄下任何新建的檔案（目前還不支援巢狀目錄）。注意：

  • 各個檔案資料格式必須一致。
  • dataDirectory中的檔案必須通過moving或者renaming來建立。
  • 一旦檔案move進dataDirectory之後，就不能再改動。所以如果這個檔案後續還有寫入，這些新寫入的資料不會被讀取。

  對於簡單的文字檔案，更簡單的方式是呼叫 streamingContext.textFileStream(dataDirectory)。

  另外，檔案資料流不是基於接收器的，所以不需要為其單獨分配一個CPU core。

  Python API fileStream目前暫時不可用，Python目前只支援textFileStream。

• 基於自定義Actor的資料流（Streams based on Custom Actors）: DStream可以由Akka actor建立得到，只需呼叫 streamingContext.actorStream(actorProps, actor-name)。詳見自定義接收器（Custom Receiver Guide）。actorStream暫時不支援Python API。
• RDD佇列資料流（Queue of RDDs as a Stream）: 如果需要測試Spark Streaming應用，你可以建立一個基於一批RDD的DStream物件，只需呼叫 streamingContext.queueStream(queueOfRDDs)。RDD會被一個個依次推入佇列，而DStream則會依次以資料流形式處理這些RDD的資料。

高階資料來源

Python API 自 Spark 1.6.1 起，Kafka、Kinesis、Flume和MQTT這些資料來源將支援Python。

使用這類資料來源需要依賴一些額外的程式碼庫，有些依賴還挺複雜的（如：Kafka、Flume）。因此為了減少依賴項版本衝突問題，各個資料來源DStream的相關功能被分割到不同的程式碼包中，只有用到的時候才需要連結打包進來。例如，如果你需要使用Twitter的tweets作為資料來源，你需要以下步驟：

1. Linking: 將spark-streaming-twitter_2.10工件加入到SBT/Maven專案依賴中。
2. Programming: 匯入TwitterUtils class，然後呼叫 TwitterUtils.createStream 建立一個DStream，具體程式碼見下放。
3. Deploying: 生成一個uber Jar包，幷包含其所有依賴項（包括 spark-streaming-twitter_2.10及其自身的依賴樹），再部署這個Jar包。部署詳情請參考部署這一節（Deploying section）。

• Scala
• Java

import org.apache.spark.streaming.twitter._

TwitterUtils.createStream(ssc, None)

注意，高階資料來源在spark-shell中不可用，因此不能用spark-shell來測試基於高階資料來源的應用。如果真有需要的話，你需要自行下載相應資料來源的Maven工件及其依賴項，並將這些Jar包部署到spark-shell的classpath中。

下面列舉了一些高階資料來源：

• Kafka: Spark Streaming 1.6.1 可相容 Kafka 0.8.2.1。詳見Kafka Integration Guide。
• Flume: Spark Streaming 1.6.1 可相容 Flume 1.6.0 。詳見Flume Integration Guide。
• Kinesis: Spark Streaming 1.6.1 可相容 Kinesis Client Library 1.2.1。詳見Kinesis Integration Guide。
• Twitter: Spark Streaming TwitterUtils 使用Twitter4j 通過 Twitter’s Streaming API 拉取公開tweets資料流。認證資訊可以用任何Twitter4j所支援的方法（methods）。你可以獲取所有的公開資料流，當然也可以基於某些關鍵詞進行過濾。示例可以參考TwitterPopularTags 和 TwitterAlgebirdCMS。

自定義資料來源

Python API 自定義資料來源目前還不支援Python。

輸入DStream也可以用自定義的方式建立。你需要做的只是實現一個自定義的接收器（receiver），以便從自定義的資料來源接收資料，然後將資料推入Spark中。詳情請參考自定義接收器指南（Custom Receiver Guide）。

接收器可靠性

從可靠性角度來劃分，大致有兩種資料來源。其中，像Kafka、Flume這樣的資料來源，它們支援對所傳輸的資料進行確認。系統收到這類可靠資料來源過來的資料，然後發出確認資訊，這樣就能夠確保任何失敗情況下，都不會丟資料。因此我們可以將接收器也相應地分為兩類：

1. 可靠接收器（Reliable Receiver） – 可靠接收器會在成功接收並儲存好Spark資料副本後，向可靠資料來源傳送確認資訊。
2. 不可靠接收器（Unreliable Receiver） – 不可靠接收器不會發送任何確認資訊。不過這種接收器常用語於不支援確認的資料來源，或者不想引入資料確認的複雜性的資料來源。

DStream支援的transformation運算元

和RDD類似，DStream也支援從輸入DStream經過各種transformation運算元對映成新的DStream。DStream支援很多RDD上常見的transformation運算元，一些常用的見下表：

下面我們會挑幾個transformation運算元深入討論一下。

updateStateByKey運算元

updateStateByKey 運算元支援維護一個任意的狀態。要實現這一點，只需要兩步：

1. 定義狀態 – 狀態資料可以是任意型別。
2. 定義狀態更新函式 – 定義好一個函式，其輸入為資料流之前的狀態和新的資料流資料，且可其更新步驟1中定義的輸入資料流的狀態。

在每一個批次資料到達後，Spark都會呼叫狀態更新函式，來更新所有已有key（不管key是否存在於本批次中）的狀態。如果狀態更新函式返回None，則對應的鍵值對會被刪除。

舉例如下。假設你需要維護一個流式應用，統計資料流中每個單詞的出現次數。這裡將各個單詞的出現次數這個整型數定義為狀態。我們接下來定義狀態更新函式如下：

• Scala
• Java

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = ...  // 將新的計數值和之前的狀態值相加，得到新的計數值
    Some(newCount)
}

該狀態更新函式可以作用於一個包括(word, 1) 鍵值對的DStream上（見本文開頭的小栗子）。

val runningCounts = pairs.updateStateByKey[Int](updateFunction _)

該狀態更新函式會為每個單詞呼叫一次，且相應的newValues是一個包含很多個”1″的陣列（這些1來自於(word,1)鍵值對），而runningCount包含之前該單詞的計數。本例的完整程式碼請參考 StatefulNetworkWordCount.scala。

注意，呼叫updateStateByKey前需要配置檢查點目錄，後續對此有詳細的討論，見檢查點（checkpointing）這節。

transform運算元

transform運算元（及其變體transformWith）可以支援任意的RDD到RDD的對映操作。也就是說，你可以用tranform運算元來包裝任何DStream API所不支援的RDD運算元。例如，將DStream每個批次中的RDD和另一個Dataset進行關聯（join）操作，這個功能DStream API並沒有直接支援。不過你可以用transform來實現這個功能，可見transform其實為DStream提供了非常強大的功能支援。比如說，你可以用事先算好的垃圾資訊，對DStream進行實時過濾。

• Scala
• Java

val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // 包含垃圾資訊的RDD

val cleanedDStream = wordCounts.transform(rdd => {
  rdd.join(spamInfoRDD).filter(...) // 將DStream中的RDD和spamInfoRDD關聯，並實時過濾垃圾資料
  ...
})

注意，這裡transform包含的運算元，其呼叫時間間隔和批次間隔是相同的。所以你可以基於時間改變對RDD的操作，如：在不同批次，呼叫不同的RDD運算元，設定不同的RDD分割槽或者廣播變數等。

基於視窗（window）的運算元

Spark Streaming同樣也提供基於時間視窗的計算，也就是說，你可以對某一個滑動時間窗內的資料施加特定tranformation運算元。如下圖所示：

如上圖所示，每次視窗滑動時，源DStream中落入視窗的RDDs就會被合併成新的windowed DStream。在上圖的例子中，這個操作會施加於3個RDD單元，而滑動距離是2個RDD單元。由此可以得出任何視窗相關操作都需要指定一下兩個引數：

• （視窗長度）window length – 視窗覆蓋的時間長度（上圖中為3）
• （滑動距離）sliding interval – 視窗啟動的時間間隔（上圖中為2）

注意，這兩個引數都必須是DStream批次間隔（上圖中為1）的整數倍.

下面咱們舉個栗子。假設，你需要擴充套件前面的那個小栗子，你需要每隔10秒統計一下前30秒內的單詞計數。為此，我們需要在包含(word, 1)鍵值對的DStream上，對最近30秒的資料呼叫reduceByKey運算元。不過這些都可以簡單地用一個 reduceByKeyAndWindow搞定。

• Scala
• Java

// 每隔10秒歸約一次最近30秒的資料
val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

以下列出了常用的視窗運算元。所有這些運算元都有前面提到的那兩個引數 – 視窗長度 和 滑動距離。

Join相關運算元

最後，值得一提的是，你在Spark Streaming中做各種關聯（join）操作非常簡單。

流-流（Stream-stream）關聯

一個數據流可以和另一個數據流直接關聯。

• Scala
• Java

val stream1: DStream[String, String] = ...
val stream2: DStream[String, String] = ...
val joinedStream = stream1.join(stream2)

上面程式碼中，stream1的每個批次中的RDD會和stream2相應批次中的RDD進行join。同樣，你可以類似地使用 leftOuterJoin, rightOuterJoin, fullOuterJoin 等。此外，你還可以基於視窗來join不同的資料流，其實現也很簡單，如下；）

• Scala
• Java

val windowedStream1 = stream1.window(Seconds(20))
val windowedStream2 = stream2.window(Minutes(1))
val joinedStream = windowedStream1.join(windowedStream2)

流-資料集（stream-dataset）關聯

其實這種情況已經在前面的DStream.transform運算元中介紹過了，這裡再舉個基於滑動視窗的例子。

• Scala
• Java

val dataset: RDD[String, String] = ...
val windowedStream = stream.window(Seconds(20))...
val joinedStream = windowedStream.transform { rdd => rdd.join(dataset) }

實際上，在上面程式碼裡，你可以動態地該表join的資料集（dataset）。傳給tranform運算元的操作函式會在每個批次重新求值，所以每次該函式都會用最新的dataset值，所以不同批次間你可以改變dataset的值。

DStream輸出運算元

輸出運算元可以將DStream的資料推送到外部系統，如：資料庫或者檔案系統。因為輸出運算元會將最終完成轉換的資料輸出到外部系統，因此只有輸出運算元呼叫時，才會真正觸發DStream transformation運算元的真正執行（這一點類似於RDD 的action運算元）。目前所支援的輸出運算元如下表：

使用foreachRDD的設計模式

DStream.foreachRDD是一個非常強大的原生工具函式，使用者可以基於此運算元將DStream資料推送到外部系統中。不過使用者需要了解如何正確而高效地使用這個工具。以下列舉了一些常見的錯誤。

通常，對外部系統寫入資料需要一些連線物件（如：遠端server的TCP連線），以便傳送資料給遠端系統。因此，開發人員可能會不經意地在Spark驅動器（driver）程序中建立一個連線物件，然後又試圖在Spark worker節點上使用這個連線。如下例所示：

dstream.foreachRDD { rdd =>
  val connection = createNewConnection()  // 這行在驅動器（driver）程序執行
  rdd.foreach { record =>
    connection.send(record) // 而這行將在worker節點上執行
  }
}

這段程式碼是錯誤的，因為它需要把連線物件序列化，再從驅動器節點發送到worker節點。而這些連線物件通常都是不能跨節點（機器）傳遞的。比如，連線物件通常都不能序列化，或者在另一個程序中反序列化後再次初始化（連線物件通常都需要初始化，因此從驅動節點發到worker節點後可能需要重新初始化）等。解決此類錯誤的辦法就是在worker節點上建立連線物件。

然而，有些開發人員可能會走到另一個極端 – 為每條記錄都建立一個連線物件，例如：

dstream.foreachRDD { rdd =>
  rdd.foreach { record =>
    val connection = createNewConnection()
    connection.send(record)
    connection.close()
  }
}

一般來說，連線物件是有時間和資源開銷限制的。因此，對每條記錄都進行一次連線物件的建立和銷燬會增加很多不必要的開銷，同時也大大減小了系統的吞吐量。一個比較好的解決方案是使用 rdd.foreachPartition – 為RDD的每個分割槽建立一個單獨的連線物件，示例如下：

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    val connection = createNewConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    connection.close()
  }
}

這樣一來，連線物件的建立開銷就攤到很多條記錄上了。

最後，還有一個更優化的辦法，就是在多個RDD批次之間複用連線物件。開發者可以維護一個靜態連線池來儲存連線物件，以便在不同批次的多個RDD之間共享同一組連線物件，示例如下：

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool 是一個靜態的、懶惰初始化的連線池
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // 將連線返還給連線池，以便後續複用之
  }
}

注意，連線池中的連線應該是懶惰建立的，並且有確定的超時時間，超時後自動銷燬。這個實現應該是目前傳送資料最高效的實現方式。

其他要點:

• DStream的轉化執行也是懶惰的，需要輸出運算元來觸發，這一點和RDD的懶惰執行由action運算元觸發很類似。特別地，DStream輸出運算元中包含的RDD action運算元會強制觸發對所接收資料的處理。因此，如果你的Streaming應用中沒有輸出運算元，或者你用了dstream.foreachRDD(func)卻沒有在func中呼叫RDD action運算元，那麼這個應用只會接收資料，而不會處理資料，接收到的資料最後只是被簡單地丟棄掉了。
• 預設地，輸出運算元只能一次執行一個，且按照它們在應用程式程式碼中定義的順序執行。

累加器和廣播變數

首先需要注意的是，累加器（Accumulators）和廣播變數（Broadcast variables）是無法從Spark Streaming的檢查點中恢復回來的。所以如果你開啟了檢查點功能，並同時在使用累加器和廣播變數，那麼你最好是使用懶惰例項化的單例模式，因為這樣累加器和廣播變數才能在驅動器（driver）故障恢復後重新例項化。程式碼示例如下：

• Scala
• Java

object WordBlacklist {

  @volatile private var instance: Broadcast[Seq[String]] = null

  def getInstance(sc: SparkContext): Broadcast[Seq[String]] = {
    if (instance == null) {
      synchronized {
        if (instance == null) {
          val wordBlacklist = Seq("a", "b", "c")
          instance = sc.broadcast(wordBlacklist)
        }
      }
    }
    instance
  }
}

object DroppedWordsCounter {

  @volatile private var instance: Accumulator[Long] = null

  def getInstance(sc: SparkContext): Accumulator[Long] = {
    if (instance == null) {
      synchronized {
        if (instance == null) {
          instance = sc.accumulator(0L, "WordsInBlacklistCounter")
        }
      }
    }
    instance
  }
}

wordCounts.foreachRDD((rdd: RDD[(String, Int)], time: Time) => {
  // 獲取現有或註冊新的blacklist廣播變數
  val blacklist = WordBlacklist.getInstance(rdd.sparkContext)
  // 獲取現有或註冊新的 droppedWordsCounter 累加器
  val droppedWordsCounter = DroppedWordsCounter.getInstance(rdd.sparkContext)
  // 基於blacklist來過濾詞，並將過濾掉的詞的個數累加到 droppedWordsCounter 中
  val counts = rdd.filter { case (word, count) =>
    if (blacklist.value.contains(word)) {
      droppedWordsCounter += count
      false
    } else {
      true
    }
  }.collect()
  val output = "Counts at time " + time + " " + counts
})

DataFrame和SQL相關運算元

在Streaming應用中可以呼叫DataFrames and SQL來處理流式資料。開發者可以用通過StreamingContext中的SparkContext物件來建立一個SQLContext，並且，開發者需要確保一旦驅動器（driver）故障恢復後，該SQLContext物件能重新創建出來。同樣，你還是可以使用懶惰建立的單例模式來例項化SQLContext，如下面的程式碼所示，這裡我們將最開始的那個小栗子做了一些修改，使用DataFrame和SQL來統計單詞計數。其實就是，將每個RDD都轉化成一個DataFrame，然後註冊成臨時表，再用SQL查詢這些臨時表。

• Scala
• Java

/** streaming應用中呼叫DataFrame運算元 */

val words: DStream[String] = ...

words.foreachRDD { rdd =>

  // 獲得SQLContext單例
  val sqlContext = SQLContext.getOrCreate(rdd.sparkContext)
  import sqlContext.implicits._

  // 將RDD[String] 轉為 DataFrame
  val wordsDataFrame = rdd.toDF("word")

  // DataFrame註冊為臨時表
  wordsDataFrame.registerTempTable("words")

  // 再用SQL語句查詢，並打印出來
  val wordCountsDataFrame = 
    sqlContext.sql("select word, count(*) as total from words group by word")
  wordCountsDataFrame.show()
}

你也可以在其他執行緒裡執行SQL查詢（非同步查詢，即：執行SQL查詢的執行緒和執行StreamingContext的執行緒不同）。不過這種情況下，你需要確保查詢的時候 StreamingContext 沒有把所需的資料丟棄掉，否則StreamingContext有可能已將老的RDD資料丟棄掉了，那麼非同步查詢的SQL語句也可能無法得到查詢結果。舉個栗子，如果你需要查詢上一個批次的資料，但是你的SQL查詢可能要執行5分鐘，那麼你就需要StreamingContext至少保留最近5分鐘的資料：streamingContext.remember(Minutes(5)) （這是Scala為例，其他語言差不多）

MLlib運算元

MLlib 提供了很多機器學習演算法。首先，你需要關注的是流式計算相關的機器學習演算法（如：Streaming Linear Regression, Streaming KMeans），這些流式演算法可以在流式資料上一邊學習訓練模型，一邊用最新的模型處理資料。除此以外，對更多的機器學習演算法而言，你需要離線訓練這些模型，然後將訓練好的模型用於線上的流式資料。詳見MLlib。

快取/持久化

和RDD類似，DStream也支援將資料持久化到記憶體中。只需要呼叫 DStream的persist() 方法，該方法內部會自動呼叫DStream中每個RDD的persist方法進而將資料持久化到記憶體中。這對於可能需要計算很多次的DStream非常有用（例如：對於同一個批資料呼叫多個運算元）。對於基於滑動視窗的運算元，如：reduceByWindow和reduceByKeyAndWindow，或者有狀態的運算元，如：updateStateByKey，資料持久化就更重要了。因此，滑動視窗運算元產生的DStream物件預設會自動持久化到記憶體中（不需要開發者呼叫persist）。

對於從網路接收資料的輸入資料流（如：Kafka、Flume、socket等），預設的持久化級別會將資料持久化到兩個不同的節點上互為備份副本，以便支援容錯。

注意，與RDD不同的是，DStream的預設持久化級別是將資料序列化到記憶體中。進一步的討論見效能調優這一小節。關於持久化級別（或者儲存級別）的更詳細說明見Spark程式設計指南（Spark Programming Guide）。

檢查點

一般來說Streaming 應用都需要7*24小時長期執行，所以必須對一些與業務邏輯無關的故障有很好的容錯（如：系統故障、JVM崩潰等）。對於這些可能性，Spark Streaming 必須在檢查點儲存足夠的資訊到一些可容錯的外部儲存系統中，以便能夠隨時從故障中恢復回來。所以，檢查點需要儲存以下兩種資料：

• 元資料檢查點（Metadata checkpointing） – 儲存流式計算邏輯的定義資訊到外部可容錯儲存系統（如：HDFS）。主要用途是用於在故障後回覆應用程式本身（後續詳談）。元數包括：
  • Configuration – 建立Streaming應用程式的配置資訊。
  • DStream operations – 定義流式處理邏輯的DStream操作資訊。
  • Incomplete batches – 已經排隊但未處理完的批次資訊。
• 資料檢查點（Data checkpointing） – 將生成的RDD儲存到可靠的儲存中。這對一些需要跨批次組合資料或者有狀態的運算元來說很有必要。在這種轉換運算元中，往往新生成的RDD是依賴於前幾個批次的RDD，因此隨著時間的推移，有可能產生很長的依賴鏈條。為了避免在恢復資料的時候需要恢復整個依賴鏈條上所有的資料，檢查點需要週期性地儲存一些中間RDD狀態資訊，以斬斷無限制增長的依賴鏈條和恢復時間。

總之，元資料檢查點主要是為了恢復驅動器節點上的故障，而資料或RDD檢查點是為了支援對有狀態轉換操作的恢復。

何時啟用檢查點

如果有以下情況出現，你就必須啟用檢查點了：

• 使用了有狀態的轉換運算元（Usage of stateful transformations） – 不管是用了 updateStateByKey 還是用了 reduceByKeyAndWindow（有”反歸約”函式的那個版本），你都必須配置檢查點目錄來週期性地儲存RDD檢查點。
• 支援驅動器故障中恢復（Recovering from failures of the driver running the application） – 這時候需要元資料檢查點以便恢復流式處理的進度資訊。

注意，一些簡單的流式應用，如果沒有用到前面所說的有狀態轉換運算元，則完全可以不開啟檢查點。不過這樣的話，驅動器（driver）故障恢復後，有可能會丟失部分資料（有些已經接收但還未處理的資料可能會丟失）。不過通常這點丟失時可接受的，很多Spark Streaming應用也是這樣執行的。對非Hadoop環境的支援未來還會繼續改進。

如何配置檢查點

檢查點的啟用，只需要設定好儲存檢查點資訊的檢查點目錄即可，一般會會將這個目錄設為一些可容錯的、可靠性較高的檔案系統（如：HDFS、S3等）。開發者只需要呼叫 streamingContext.checkpoint(checkpointDirectory)。設定好檢查點，你就可以使用前面提到的有狀態轉換運算元了。另外，如果你需要你的應用能夠支援從驅動器故障中恢復，你可能需要重寫部分程式碼，實現以下行為：

• 如果程式是首次啟動，就需要new一個新的StreamingContext，並定義好所有的資料流處理，然後呼叫StreamingContext.start()。
• 如果程式是故障後重啟，就需要從檢查點目錄中的資料中重新構建StreamingContext物件。

• Scala
• Java

不過這個行為可以用StreamingContext.getOrCreate來實現，示例如下：

// 首次建立StreamingContext並定義好資料流處理邏輯
def functionToCreateContext(): StreamingContext = {
    val ssc = new StreamingContext(...)   // 新建一個StreamingContext物件
    val lines = ssc.socketTextStream(...) // 建立DStreams
    ...
    ssc.checkpoint(checkpointDirectory)   // 設定好檢查點目錄
    ssc
}

// 建立新的StreamingContext物件，或者從檢查點構造一個
val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)

// 無論是否是首次啟動都需要設定的工作在這裡
context. ...

// 啟動StreamingContext物件
context.start()
context.awaitTermination()

如果 checkpointDirectory 目錄存在，則context物件會從檢查點資料重新構建出來。如果該目錄不存在（如：首次執行），則 functionToCreateContext 函式會被呼叫，建立一個新的StreamingContext物件並定義好DStream資料流。完整的示例請參見RecoverableNetworkWordCount，這個例子會將網路資料中的單詞計數統計結果新增到一個檔案中。

除了使用getOrCreate之外，開發者還需要確保驅動器程序能在故障後重啟。這一點只能由應用的部署環境基礎設施來保證。進一步的討論見部署（Deployment）這一節。

另外需要注意的是，RDD檢查點會增加額外的儲存資料的開銷。這可能會導致資料流的處理時間變長。因此，你必須仔細的調整檢查點間隔時間。如果批次間隔太小（比如：1秒），那麼對每個批次儲存檢查點資料將大大減小吞吐量。另一方面，檢查點儲存過於頻繁又會導致血統資訊和任務個數的增加，這同樣會影響系統性能。對於需要RDD檢查點的有狀態轉換運算元，預設的間隔是批次間隔的整數倍，且最小10秒。開發人員可以這樣來自定義這個間隔：dstream.checkpoint(checkpointInterval)。一般推薦設為批次間隔時間的5~10倍。

部署應用

本節中將主要討論一下如何部署Spark Streaming應用。

前提條件

要執行一個Spark Streaming 應用，你首先需要具備以下條件：

• 叢集以及叢集管理器 – 這是一般Spark應用的基本要求，詳見 deployment guide。
• 給Spark應用打個JAR包 – 你需要將你的應用打成一個JAR包。如果使用spark-submit 提交應用，那麼你不需要提供Spark和Spark Streaming的相關JAR包。但是，如果你使用了高階資料來源（advanced sources – 如：Kafka、Flume、Twitter等），那麼你需要將這些高階資料來源相關的JAR包及其依賴一起打包並部署。例如，如果你使用了TwitterUtils，那麼就必須將spark-streaming-twitter_2.10及其相關依賴都打到應用的JAR包中。
• 為執行器（executor）預留足夠的記憶體 – 執行器必須配置預留好足夠的記憶體，因為接受到的資料都得存在記憶體裡。注意，如果某些視窗長度達到10分鐘，那也就是說你的系統必須知道保留10分鐘的資料在記憶體裡。可見，到底預留多少記憶體是取決於你的應用處理邏輯的。
• 配置檢查點 – 如果你的流式應用需要檢查點，那麼你需要配置一個Hadoop API相容的可容錯儲存目錄作為檢查點目錄，流式應用的資訊會寫入這個目錄，故障恢復時會用到這個目錄下的資料。詳見前面的檢查點小節。
• 配置驅動程式自動重啟 – 流式應用自動恢復的前提就是，部署基礎設施能夠監控驅動器程序，並且能夠在其故障時，自動重啟之。不同的叢集管理器有不同的工具來實現這一功能：
  • Spark獨立部署 – Spark獨立部署叢集可以支援將Spark應用的驅動器提交到叢集的某個worker節點上執行。同時，Spark的叢集管理器可以對該驅動器程序進行監控，一旦驅動器退出且返回非0值，或者因worker節點原始失敗，Spark叢集管理器將自動重啟這個驅動器。詳見Spark獨立部署指南（Spark Standalone guide）。
  • YARN – YARN支援和獨立部署類似的重啟機制。詳細請參考YARN的文件。
  • Mesos – Mesos上需要用Marathon來實現這一功能。
• 配置WAL（write ahead log）- 從Spark 1.2起，我們引入了write ahead log來提高容錯性。如果啟用這個功能，則所有接收到的資料都會以write ahead log形式寫入配置好的檢查點目錄中。這樣就能確保資料零丟失（容錯語義有詳細的討論）。使用者只需將 spark.streaming.receiver.writeAheadLog 設為true。不過，這同樣可能會導致接收器的吞吐量下降。不過你可以啟動多個接收器並行接收資料，從而提升整體的吞吐量（more receivers in parallel）。另外，建議在啟用WAL後禁用掉接收資料多副本功能，因為WAL其實已經是儲存在一個多副本儲存系統中了。你只需要把儲存級別設為 StorageLevel.MEMORY_AND_DISK_SER。如果是使用S3（或者其他不支援flushing的檔案系統）儲存WAL，一定要記得啟用這兩個標識：spark.streaming.driver.writeAheadLog.closeFileAfterWrite 和 spark.streaming.receiver.writeAheadLog.closeFileAfterWrite。更詳細請參考： Spark Streaming Configuration。
• 設定好最大接收速率 – 如果叢集可用資源不足以跟上接收資料的速度，那麼可以在接收器設定一下最大接收速率，即：每秒接收記錄的條數。相關的主要配置有：spark.streaming.receiver.maxRate，如果使用Kafka Direct API 還需要設定 spark.streaming.kafka.maxRatePerPartition。從Spark 1.5起，我們引入了backpressure的概念來動態地根據叢集處理速度，評估並調整該接收速率。使用者只需將 spark.streaming.backpressure.enabled設為true即可啟用該功能。

升級應用程式碼

升級Spark Streaming應用程式程式碼，可以使用以下兩種方式：

• 新的Streaming程式和老的並行跑一段時間，新程式完成初始化以後，再關閉老的。注意，這種方式適用於能同時傳送資料到多個目標的資料來源（即：資料來源同時將資料發給新老兩個Streaming應用程式）。
• 老程式能夠優雅地退出（參考  StreamingContext.stop(...) or JavaStreamingContext.stop(...) ），即：確保所收到的資料都已經處理完畢後再退出。然後再啟動新的Streaming程式，而新程式將接著在老程式退出點上繼續拉取資料。注意，這種方式需要資料來源支援資料快取（或者叫資料堆積，如：Kafka、Flume），因為在新舊程式交接的這個空檔時間，資料需要在資料來源處快取。目前還不能支援從檢查點重啟，因為檢查點儲存的資訊包含老程式中的序列化物件資訊，在新程式中將其反序列化可能會出錯。這種情況下，只能要麼指定一個新的檢查點目錄，要麼刪除老的檢查點目錄。

應用監控

除了Spark自身的監控能力（monitoring capabilities）之外，對Spark Streaming還有一些額外的監控功能可用。如果例項化了StreamingContext，那麼你可以在Spark web UI上看到多出了一個Streaming tab頁，上面顯示了正在執行的接收器（是否活躍，接收記錄的條數，失敗資訊等）和處理完的批次資訊（批次處理時間，查詢延時等）。這些資訊都可以用來監控streaming應用。

web UI上有兩個度量特別重要：

• 批次處理耗時（Processing Time） – 處理單個批次耗時
• 批次排程延時（Scheduling Delay） -各批次在佇列中等待時間（等待上一個批次處理完）

如果批次處理耗時一直比批次間隔時間大，或者批次排程延時持續上升，就意味著系統處理速度跟不上資料接收速度。這時候你就得考慮一下怎麼把批次處理時間降下來（reducing）。

Spark Streaming程式的處理進度可以用StreamingListener介面來監聽，這個介面可以監聽到接收器的狀態和處理時間。不過需要注意的是，這是一個developer API介