摘要：輸入和接收器輸入代表從某種流式數(shù)據(jù)源流入的數(shù)據(jù)流。文件數(shù)據(jù)流可以從任何兼容包括等的文件系統(tǒng)，創(chuàng)建方式如下將監(jiān)視該目錄，并處理該目錄下任何新建的文件目前還不支持嵌套目錄。會(huì)被一個(gè)個(gè)依次推入隊(duì)列，而則會(huì)依次以數(shù)據(jù)流形式處理這些的數(shù)據(jù)。

特點(diǎn)：

Spark Streaming能夠?qū)崿F(xiàn)對(duì)實(shí)時(shí)數(shù)據(jù)流的流式處理，并具有很好的可擴(kuò)展性、高吞吐量和容錯(cuò)性。

Spark Streaming支持從多種數(shù)據(jù)源提取數(shù)據(jù)，如：Kafka、Flume、Twitter、ZeroMQ、Kinesis以及TCP套接字，并且可以提供一些高級(jí)API來(lái)表達(dá)復(fù)雜的處理算法，如：map、reduce、join和window等。

Spark Streaming支持將處理完的數(shù)據(jù)推送到文件系統(tǒng)、數(shù)據(jù)庫(kù)或者實(shí)時(shí)儀表盤中展示。

可以將Spark的機(jī)器學(xué)習(xí)（machine learning） 和 圖計(jì)算（graph processing）的算法應(yīng)用于Spark Streaming的數(shù)據(jù)流當(dāng)中。

下圖展示了Spark Streaming的內(nèi)部工作原理。Spark Streaming從實(shí)時(shí)數(shù)據(jù)流接入數(shù)據(jù)，再將其劃分為一個(gè)個(gè)小批量供后續(xù)Spark engine處理，所以實(shí)際上，Spark Streaming是按一個(gè)個(gè)小批量來(lái)處理數(shù)據(jù)流的。

Spark Streaming為這種持續(xù)的數(shù)據(jù)流提供了的一個(gè)高級(jí)抽象，即：discretized stream（離散數(shù)據(jù)流）或者叫DStream。DStream既可以從輸入數(shù)據(jù)源創(chuàng)建得來(lái)，如：Kafka、Flume或者Kinesis，也可以從其他DStream經(jīng)一些算子操作得到。其實(shí)在內(nèi)部，一個(gè)DStream就是包含了一系列RDDs。

        SparkConf conf = new SparkConf().setAppName("stream1").setMaster("local[2]");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(1));
        JavaReceiverInputDStream lines = jsc.socketTextStream("localhost", 9999);
         JavaPairDStream pairs=
                 lines.flatMap((str)->Arrays.asList(str.split(" ")).iterator())
                 .mapToPair((str)->new Tuple2(str,1L));
         JavaPairDStream res=pairs.reduceByKey((v1,v2)->v1+v2);
         res.print();
         
        jsc.start();
        try {
            jsc.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

StreamingContext 是Spark Streaming的入口。并將批次間隔設(shè)為1秒。
利用這個(gè)上下文對(duì)象（StreamingContext），我們可以創(chuàng)建一個(gè)DStream，該DStream代表從前面的TCP數(shù)據(jù)源流入的數(shù)據(jù)流，同時(shí)TCP數(shù)據(jù)源是由主機(jī)名（如：hostnam）和端口（如：9999）來(lái)描述的。
這里的 lines 就是從數(shù)據(jù)server接收到的數(shù)據(jù)流。其中每一條記錄都是一行文本。接下來(lái)，我們就需要把這些文本行按空格分割成單詞。
flatMap 是一種 “一到多”（one-to-many）的映射算子，它可以將源DStream中每一條記錄映射成多條記錄，從而產(chǎn)生一個(gè)新的DStream對(duì)象。在本例中，lines中的每一行都會(huì)被flatMap映射為多個(gè)單詞，從而生成新的words DStream對(duì)象。然后，我們就能對(duì)這些單詞進(jìn)行計(jì)數(shù)了。
words這個(gè)DStream對(duì)象經(jīng)過(guò)map算子（一到一的映射）轉(zhuǎn)換為一個(gè)包含（word, 1）鍵值對(duì)的DStream對(duì)象pairs，再對(duì)pairs使用reduce算子，得到每個(gè)批次中各個(gè)單詞的出現(xiàn)頻率。
注意，執(zhí)行以上代碼后，Spark Streaming只是將計(jì)算邏輯設(shè)置好，此時(shí)并未真正的開(kāi)始處理數(shù)據(jù)。要啟動(dòng)之前的處理邏輯，我們還需要如下調(diào)用：

ssc.start()             // 啟動(dòng)流式計(jì)算
ssc.awaitTermination()  // 等待直到計(jì)算終止

首先，你需要運(yùn)行netcat（Unix-like系統(tǒng)都會(huì)有這個(gè)小工具），將其作為data server

$ nc -lk 9999

然后,執(zhí)行程序. 現(xiàn)在你嘗試可以在運(yùn)行netcat的終端里敲幾個(gè)單詞，你會(huì)發(fā)現(xiàn)這些單詞以及相應(yīng)的計(jì)數(shù)會(huì)出現(xiàn)在啟動(dòng)Spark Streaming例子的終端屏幕上。

注意，StreamingContext在內(nèi)部會(huì)創(chuàng)建一個(gè) SparkContext 對(duì)象（SparkContext是所有Spark應(yīng)用的入口，在StreamingContext對(duì)象中可以這樣訪問(wèn)：ssc.sparkContext）。

StreamingContext還有另一個(gè)構(gòu)造參數(shù)，即：批次間隔，這個(gè)值的大小需要根據(jù)應(yīng)用的具體需求和可用的集群資源來(lái)確定。

需要關(guān)注的重點(diǎn):

一旦streamingContext啟動(dòng)，就不能再對(duì)其計(jì)算邏輯進(jìn)行添加或修改。

一旦streamingContext被stop掉，就不能restart。

單個(gè)JVM虛機(jī)同一時(shí)間只能包含一個(gè)active的StreamingContext。

StreamingContext.stop() 也會(huì)把關(guān)聯(lián)的SparkContext對(duì)象stop掉，如果不想把SparkContext對(duì)象也stop掉，可以將StreamingContext.stop的可選參數(shù) stopSparkContext 設(shè)為false。

一個(gè)SparkContext對(duì)象可以和多個(gè)StreamingContext對(duì)象關(guān)聯(lián)，只要先對(duì)前一個(gè)StreamingContext.stop(sparkContext=false)，然后再創(chuàng)建新的StreamingContext對(duì)象即可。

離散數(shù)據(jù)流（DStream）是Spark Streaming最基本的抽象。它代表了一種連續(xù)的數(shù)據(jù)流，要么從某種數(shù)據(jù)源提取數(shù)據(jù)，要么從其他數(shù)據(jù)流映射轉(zhuǎn)換而來(lái)。DStream內(nèi)部是由一系列連續(xù)的RDD組成的，每個(gè)RDD都包含了特定時(shí)間間隔內(nèi)的一批數(shù)據(jù)，如下圖所示：

任何作用于DStream的算子，其實(shí)都會(huì)被轉(zhuǎn)化為對(duì)其內(nèi)部RDD的操作。底層的RDD轉(zhuǎn)換仍然是由Spark引擎來(lái)計(jì)算。DStream的算子將這些細(xì)節(jié)隱藏了起來(lái)，并為開(kāi)發(fā)者提供了更為方便的高級(jí)API。

輸入DStream代表從某種流式數(shù)據(jù)源流入的數(shù)據(jù)流。在之前的例子里，lines 對(duì)象就是輸入DStream，它代表從netcat server收到的數(shù)據(jù)流。每個(gè)輸入DStream（除文件數(shù)據(jù)流外）都和一個(gè)接收器（Receiver）相關(guān)聯(lián)，而接收器則是專門從數(shù)據(jù)源拉取數(shù)據(jù)到內(nèi)存中的對(duì)象。

Spark Streaming主要提供兩種內(nèi)建的流式數(shù)據(jù)源：

基礎(chǔ)數(shù)據(jù)源（Basic sources）: 在StreamingContext API 中可直接使用的源，如：文件系統(tǒng)，套接字連接或者Akka actor。

高級(jí)數(shù)據(jù)源（Advanced sources）: 需要依賴額外工具類的源，如：Kafka、Flume、Kinesis、Twitter等數(shù)據(jù)源。這些數(shù)據(jù)源都需要增加額外的依賴，詳見(jiàn)依賴鏈接（linking）這一節(jié)。

注意，如果你需要同時(shí)從多個(gè)數(shù)據(jù)源拉取數(shù)據(jù)，那么你就需要?jiǎng)?chuàng)建多個(gè)DStream對(duì)象。多個(gè)DStream對(duì)象其實(shí)也就同時(shí)創(chuàng)建了多個(gè)數(shù)據(jù)流接收器。但是請(qǐng)注意，Spark的worker/executor 都是長(zhǎng)期運(yùn)行的，因此它們都會(huì)各自占用一個(gè)分配給Spark Streaming應(yīng)用的CPU。
因此，本地運(yùn)行時(shí)，一定要將master設(shè)為”local[n]”，其中 n > 接收器的個(gè)數(shù)。
將Spark Streaming應(yīng)用置于集群中運(yùn)行時(shí)，同樣，分配給該應(yīng)用的CPU core數(shù)必須大于接收器的總數(shù)。否則，該應(yīng)用就只會(huì)接收數(shù)據(jù)，而不會(huì)處理數(shù)據(jù)。

使用ssc.socketTextStream(…) 可以從一個(gè)TCP連接中接收文本數(shù)據(jù)。而除了TCP套接字外,StreamingContext API 還支持從文件或者Akka actor中拉取數(shù)據(jù)。
文件數(shù)據(jù)流（File Streams）: 可以從任何兼容HDFS API（包括：HDFS、S3、NFS等）的文件系統(tǒng)，創(chuàng)建方式如下：

streamingContext.fileStream(dataDirectory);

Spark Streaming將監(jiān)視該dataDirectory目錄，并處理該目錄下任何新建的文件（目前還不支持嵌套目錄）。注意：

各個(gè)文件數(shù)據(jù)格式必須一致。

dataDirectory中的文件必須通過(guò)moving或者renaming來(lái)創(chuàng)建。

一旦文件move進(jìn)dataDirectory之后，就不能再改動(dòng)。所以如果這個(gè)文件后續(xù)還有寫入，這些新寫入的數(shù)據(jù)不會(huì)被讀取。

對(duì)于簡(jiǎn)單的文本文件，更簡(jiǎn)單的方式是調(diào)用 streamingContext.textFileStream(dataDirectory)。

另外，文件數(shù)據(jù)流不是基于接收器的，所以不需要為其多帶帶分配一個(gè)CPU core。

RDD隊(duì)列數(shù)據(jù)流（Queue of RDDs as a Stream）: 如果需要測(cè)試Spark Streaming應(yīng)用，你可以創(chuàng)建一個(gè)基于一批RDD的DStream對(duì)象，只需調(diào)用 streamingContext.queueStream(queueOfRDDs)。RDD會(huì)被一個(gè)個(gè)依次推入隊(duì)列，而DStream則會(huì)依次以數(shù)據(jù)流形式處理這些RDD的數(shù)據(jù)。

自定義數(shù)據(jù)源
輸入DStream也可以用自定義的方式創(chuàng)建。你需要做的只是實(shí)現(xiàn)一個(gè)自定義的接收器（receiver），以便從自定義的數(shù)據(jù)源接收數(shù)據(jù)，然后將數(shù)據(jù)推入Spark中。 見(jiàn)：http://spark.apache.org/docs/...

接收器可靠性
從可靠性角度來(lái)劃分，大致有兩種數(shù)據(jù)源。其中，像Kafka、Flume這樣的數(shù)據(jù)源，它們支持對(duì)所傳輸?shù)臄?shù)據(jù)進(jìn)行確認(rèn)。系統(tǒng)收到這類可靠數(shù)據(jù)源過(guò)來(lái)的數(shù)據(jù)，然后發(fā)出確認(rèn)信息，這樣就能夠確保任何失敗情況下，都不會(huì)丟數(shù)據(jù)。因此我們可以將接收器也相應(yīng)地分為兩類：

可靠接收器（Reliable Receiver） – 可靠接收器會(huì)在成功接收并保存好Spark數(shù)據(jù)副本后，向可靠數(shù)據(jù)源發(fā)送確認(rèn)信息。

不可靠接收器（Unreliable Receiver） – 不可靠接收器不會(huì)發(fā)送任何確認(rèn)信息。

和RDD類似，DStream也支持從輸入DStream經(jīng)過(guò)各種transformation算子映射成新的DStream。

map(func) 返回會(huì)一個(gè)新的DStream，并將源DStream中每個(gè)元素通過(guò)func映射為新的元素

flatMap(func) 和map類似，不過(guò)每個(gè)輸入元素不再是映射為一個(gè)輸出，而是映射為0到多個(gè)輸出

filter(func) 返回一個(gè)新的DStream，并包含源DStream中被func選中（func返回true）的元素

repartition(numPartitions) 更改DStream的并行度（增加或減少分區(qū)數(shù)）

union(otherStream) 返回新的DStream，包含源DStream和otherDStream元素的并集

count() 返回一個(gè)包含單元素RDDs的DStream，其中每個(gè)元素是源DStream中各個(gè)RDD中的元素個(gè)數(shù)

reduce(func) 返回一個(gè)包含單元素RDDs的DStream，其中每個(gè)元素是通過(guò)源RDD中各個(gè)RDD的元素經(jīng)func（func輸入兩個(gè)參數(shù)并返回一個(gè)同類型結(jié)果數(shù)據(jù)）聚合得到的結(jié)果。func必須滿足結(jié)合律，以便支持并行計(jì)算。

countByValue() 如果源DStream包含的元素類型為K，那么該算子返回新的DStream包含元素為(K, Long)鍵值對(duì)，其中K為源DStream各個(gè)元素，而Long為該元素出現(xiàn)的次數(shù)。

reduceByKey(func, [numTasks]) 如果源DStream 包含的元素為 (K, V) 鍵值對(duì)，則該算子返回一個(gè)新的也包含(K, V)鍵值對(duì)的DStream，其中V是由func聚合得到的。注意：默認(rèn)情況下，該算子使用Spark的默認(rèn)并發(fā)任務(wù)數(shù)（本地模式為2，集群模式下由spark.default.parallelism 決定）。你可以通過(guò)可選參數(shù)numTasks來(lái)指定并發(fā)任務(wù)個(gè)數(shù)。

join(otherStream, [numTasks]) 如果源DStream包含元素為(K, V)，同時(shí)otherDStream包含元素為(K, W)鍵值對(duì)，則該算子返回一個(gè)新的DStream，其中源DStream和otherDStream中每個(gè)K都對(duì)應(yīng)一個(gè) (K, (V, W))鍵值對(duì)元素。

cogroup(otherStream, [numTasks]) 如果源DStream包含元素為(K, V)，同時(shí)otherDStream包含元素為(K, W)鍵值對(duì)，則該算子返回一個(gè)新的DStream，其中每個(gè)元素類型為包含(K, Seq[V], Seq[W])的tuple。

transform(func) 返回一個(gè)新的DStream，其包含的RDD為源RDD經(jīng)過(guò)func操作后得到的結(jié)果。利用該算子可以對(duì)DStream施加任意的操作。

updateStateByKey(func) 返回一個(gè)包含新”狀態(tài)”的DStream。源DStream中每個(gè)key及其對(duì)應(yīng)的values會(huì)作為func的輸入，而func可以用于對(duì)每個(gè)key的“狀態(tài)”數(shù)據(jù)作任意的更新操作。

updateStateByKey 算子支持維護(hù)一個(gè)任意的狀態(tài)。要實(shí)現(xiàn)這一點(diǎn)，只需要兩步：

定義狀態(tài) – 狀態(tài)數(shù)據(jù)可以是任意類型。

定義狀態(tài)更新函數(shù) – 定義好一個(gè)函數(shù)，其輸入為數(shù)據(jù)流之前的狀態(tài)和新的數(shù)據(jù)流數(shù)據(jù)，且可其更新步驟1中定義的輸入數(shù)據(jù)流的狀態(tài)。
在每一個(gè)批次數(shù)據(jù)到達(dá)后，Spark都會(huì)調(diào)用狀態(tài)更新函數(shù)，來(lái)更新所有已有key（不管key是否存在于本批次中）的狀態(tài)。如果狀態(tài)更新函數(shù)返回None，則對(duì)應(yīng)的鍵值對(duì)會(huì)被刪除。

舉例如下。假設(shè)你需要維護(hù)一個(gè)流式應(yīng)用，統(tǒng)計(jì)數(shù)據(jù)流中每個(gè)單詞的出現(xiàn)次數(shù)。這里將各個(gè)單詞的出現(xiàn)次數(shù)這個(gè)整型數(shù)定義為狀態(tài)。我們接下來(lái)定義狀態(tài)更新函數(shù)如下：

Function2, Optional, Optional> updateFunction =
  new Function2, Optional, Optional>() {
    @Override public Optional call(List values, Optional state) {
      Integer newSum = ...  // add the new values with the previous running count to get the new count
      return Optional.of(newSum);
    }
  };

注意，調(diào)用updateStateByKey前需要配置檢查點(diǎn)目錄. 配置方式見(jiàn)下：

一般來(lái)說(shuō)Streaming 應(yīng)用都需要7*24小時(shí)長(zhǎng)期運(yùn)行，所以必須對(duì)一些與業(yè)務(wù)邏輯無(wú)關(guān)的故障有很好的容錯(cuò)（如：系統(tǒng)故障、JVM崩潰等）。對(duì)于這些可能性，Spark Streaming 必須在檢查點(diǎn)保存足夠的信息到一些可容錯(cuò)的外部存儲(chǔ)系統(tǒng)中，以便能夠隨時(shí)從故障中恢復(fù)回來(lái)。所以，檢查點(diǎn)需要保存以下兩種數(shù)據(jù)：

元數(shù)據(jù)檢查點(diǎn)（Metadata checkpointing） – 保存流式計(jì)算邏輯的定義信息到外部可容錯(cuò)存儲(chǔ)系統(tǒng)（如：HDFS）。主要用途是用于在故障后回復(fù)應(yīng)用程序本身（后續(xù)詳談）。元數(shù)包括：

Configuration – 創(chuàng)建Streaming應(yīng)用程序的配置信息。

DStream operations – 定義流式處理邏輯的DStream操作信息。

Incomplete batches – 已經(jīng)排隊(duì)但未處理完的批次信息。
總之，元數(shù)據(jù)檢查點(diǎn)主要是為了恢復(fù)驅(qū)動(dòng)器節(jié)點(diǎn)上的故障，而數(shù)據(jù)或RDD檢查點(diǎn)是為了支持對(duì)有狀態(tài)轉(zhuǎn)換操作的恢復(fù)。

何時(shí)啟用檢查點(diǎn)

使用了有狀態(tài)的轉(zhuǎn)換算子（Usage of stateful transformations） – 不管是用了 updateStateByKey 還是用了 reduceByKeyAndWindow（有”反歸約”函數(shù)的那個(gè)版本），你都必須配置檢查點(diǎn)目錄來(lái)周期性地保存RDD檢查點(diǎn)。

支持驅(qū)動(dòng)器故障中恢復(fù)（Recovering from failures of the driver running the application） – 這時(shí)候需要元數(shù)據(jù)檢查點(diǎn)以便恢復(fù)流式處理的進(jìn)度信息。

注意，一些簡(jiǎn)單的流式應(yīng)用，如果沒(méi)有用到前面所說(shuō)的有狀態(tài)轉(zhuǎn)換算子，則完全可以不開(kāi)啟檢查點(diǎn)。不過(guò)這樣的話，驅(qū)動(dòng)器（driver）故障恢復(fù)后，有可能會(huì)丟失部分?jǐn)?shù)據(jù)（有些已經(jīng)接收但還未處理的數(shù)據(jù)可能會(huì)丟失）。不過(guò)通常這點(diǎn)丟失時(shí)可接受的，很多Spark Streaming應(yīng)用也是這樣運(yùn)行的。

如何配置檢查點(diǎn)

檢查點(diǎn)的啟用，只需要設(shè)置好保存檢查點(diǎn)信息的檢查點(diǎn)目錄即可，一般會(huì)會(huì)將這個(gè)目錄設(shè)為一些可容錯(cuò)的、可靠性較高的文件系統(tǒng)（如：HDFS、S3等）。
第一種：開(kāi)發(fā)者只需要調(diào)用 streamingContext.checkpoint(checkpointDirectory)。設(shè)置好檢查點(diǎn)，你就可以使用前面提到的有狀態(tài)轉(zhuǎn)換算子了。

第二種：如果你需要你的應(yīng)用能夠支持從驅(qū)動(dòng)器故障中恢復(fù)，你可能需要重寫部分代碼，實(shí)現(xiàn)以下行為：

如果程序是首次啟動(dòng)，就需要new一個(gè)新的StreamingContext，并定義好所有的數(shù)據(jù)流處理，然后調(diào)用StreamingContext.start()。

如果程序是故障后重啟，就需要從檢查點(diǎn)目錄中的數(shù)據(jù)中重新構(gòu)建StreamingContext對(duì)象。

// Create a factory object that can create and setup a new JavaStreamingContext
JavaStreamingContextFactory contextFactory = new JavaStreamingContextFactory() {
  @Override public JavaStreamingContext create() {
    JavaStreamingContext jssc = new JavaStreamingContext(...);  // new context
    JavaDStream lines = jssc.socketTextStream(...);     // create DStreams
    ...
    jssc.checkpoint(checkpointDirectory);                       // set checkpoint directory
    return jssc;
  }
};

// Get JavaStreamingContext from checkpoint data or create a new one
JavaStreamingContext context = JavaStreamingContext.getOrCreate(checkpointDirectory, contextFactory);

// Do additional setup on context that needs to be done,
// irrespective of whether it is being started or restarted
context. ...

// Start the context
context.start();
context.awaitTermination();

需要注意的是，RDD檢查點(diǎn)會(huì)增加額外的保存數(shù)據(jù)的開(kāi)銷。這可能會(huì)導(dǎo)致數(shù)據(jù)流的處理時(shí)間變長(zhǎng)。

因此，你必須仔細(xì)的調(diào)整檢查點(diǎn)間隔時(shí)間。如果批次間隔太小（比如：1秒），那么對(duì)每個(gè)批次保存檢查點(diǎn)數(shù)據(jù)將大大減小吞吐量。

另一方面，檢查點(diǎn)保存過(guò)于頻繁又會(huì)導(dǎo)致血統(tǒng)信息和任務(wù)個(gè)數(shù)的增加，這同樣會(huì)影響系統(tǒng)性能。

對(duì)于需要RDD檢查點(diǎn)的有狀態(tài)轉(zhuǎn)換算子，默認(rèn)的間隔是批次間隔的整數(shù)倍，且最小10秒。開(kāi)發(fā)人員可以這樣來(lái)自定義這個(gè)間隔：dstream.checkpoint(checkpointInterval)。一般推薦設(shè)為批次間隔時(shí)間的5~10倍。

transform算子（及其變體transformWith）可以支持任意的RDD到RDD的映射操作。也就是說(shuō)，你可以用tranform算子來(lái)包裝任何DStream API所不支持的RDD算子。例如，將DStream每個(gè)批次中的RDD和另一個(gè)Dataset進(jìn)行關(guān)聯(lián)（join）操作，這個(gè)功能DStream API并沒(méi)有直接支持。不過(guò)你可以用transform來(lái)實(shí)現(xiàn)這個(gè)功能，可見(jiàn)transform其實(shí)為DStream提供了非常強(qiáng)大的功能支持。比如說(shuō)，你可以用事先算好的垃圾信息，對(duì)DStream進(jìn)行實(shí)時(shí)過(guò)濾。

// RDD containing spam information
final JavaPairRDD spamInfoRDD = jssc.sparkContext().newAPIHadoopRDD(...);

JavaPairDStream cleanedDStream = wordCounts.transform(
  new Function, JavaPairRDD>() {
    @Override public JavaPairRDD call(JavaPairRDD rdd) throws Exception {
      rdd.join(spamInfoRDD).filter(...); // join data stream with spam information to do data cleaning
      ...
    }
  });

注意，這里transform包含的算子，其調(diào)用時(shí)間間隔和批次間隔是相同的。所以你可以基于時(shí)間改變對(duì)RDD的操作，如：在不同批次，調(diào)用不同的RDD算子，設(shè)置不同的RDD分區(qū)或者廣播變量等。

Spark Streaming同樣也提供基于時(shí)間窗口的計(jì)算，也就是說(shuō)，你可以對(duì)某一個(gè)滑動(dòng)時(shí)間窗內(nèi)的數(shù)據(jù)施加特定tranformation算子。如下圖所示：

如上圖所示，每次窗口滑動(dòng)時(shí)，源DStream中落入窗口的RDDs就會(huì)被合并成新的windowed DStream。在上圖的例子中，這個(gè)操作會(huì)施加于3個(gè)RDD單元，而滑動(dòng)距離是2個(gè)RDD單元。由此可以得出任何窗口相關(guān)操作都需要指定一下兩個(gè)參數(shù)：

（窗口長(zhǎng)度）window length – 窗口覆蓋的時(shí)間長(zhǎng)度（上圖中為3）

（滑動(dòng)距離）sliding interval – 窗口啟動(dòng)的時(shí)間間隔（上圖中為2）

注意，這兩個(gè)參數(shù)都必須是DStream批次間隔（上圖中為1）的整數(shù)倍.

下面咱們舉個(gè)例子。假設(shè)，你需要每隔10秒統(tǒng)計(jì)一下前30秒內(nèi)的單詞計(jì)數(shù)。為此，我們需要在包含(word, 1)鍵值對(duì)的DStream上，對(duì)最近30秒的數(shù)據(jù)調(diào)用reduceByKey算子。不過(guò)這些都可以簡(jiǎn)單地用一個(gè) reduceByKeyAndWindow搞定。

// Reduce function adding two integers, defined separately for clarity
Function2 reduceFunc = new Function2() {
  @Override public Integer call(Integer i1, Integer i2) {
    return i1 + i2;
  }
};

// 每隔10秒歸約一次最近30秒的數(shù)據(jù)
JavaPairDStream windowedWordCounts = pairs.reduceByKeyAndWindow(reduceFunc, Durations.seconds(30), Durations.seconds(10));

以下列出了常用的窗口算子。所有這些算子都有前面提到的那兩個(gè)參數(shù) – 窗口長(zhǎng)度 和 滑動(dòng)距離。

window(windowLength, slideInterval) 將源DStream窗口化，并返回轉(zhuǎn)化后的DStream

countByWindow(windowLength,slideInterval) 返回?cái)?shù)據(jù)流在一個(gè)滑動(dòng)窗口內(nèi)的元素個(gè)數(shù)

reduceByWindow(func, windowLength,slideInterval) 基于數(shù)據(jù)流在一個(gè)滑動(dòng)窗口內(nèi)的元素，用func做聚合，返回一個(gè)單元素?cái)?shù)據(jù)流。func必須滿足結(jié)合律，以便支持并行計(jì)算。

reduceByKeyAndWindow(func,windowLength, slideInterval, [numTasks]) 基于(K, V)鍵值對(duì)DStream，將一個(gè)滑動(dòng)窗口內(nèi)的數(shù)據(jù)進(jìn)行聚合，返回一個(gè)新的包含(K,V)鍵值對(duì)的DStream，其中每個(gè)value都是各個(gè)key經(jīng)過(guò)func聚合后的結(jié)果。
注意：如果不指定numTasks，其值將使用Spark的默認(rèn)并行任務(wù)數(shù)（本地模式下為2，集群模式下由 spark.default.parallelism決定）。當(dāng)然，你也可以通過(guò)numTasks來(lái)指定任務(wù)個(gè)數(shù)。

reduceByKeyAndWindow(func, invFunc,windowLength,slideInterval, [numTasks]) 和前面的reduceByKeyAndWindow() 類似，只是這個(gè)版本會(huì)用之前滑動(dòng)窗口計(jì)算結(jié)果，遞增地計(jì)算每個(gè)窗口的歸約結(jié)果。當(dāng)新的數(shù)據(jù)進(jìn)入窗口時(shí)，這些values會(huì)被輸入func做歸約計(jì)算，而這些數(shù)據(jù)離開(kāi)窗口時(shí)，對(duì)應(yīng)的這些values又會(huì)被輸入 invFunc 做”反歸約”計(jì)算。舉個(gè)簡(jiǎn)單的例子，就是把新進(jìn)入窗口數(shù)據(jù)中各個(gè)單詞個(gè)數(shù)“增加”到各個(gè)單詞統(tǒng)計(jì)結(jié)果上，同時(shí)把離開(kāi)窗口數(shù)據(jù)中各個(gè)單詞的統(tǒng)計(jì)個(gè)數(shù)從相應(yīng)的統(tǒng)計(jì)結(jié)果中“減掉”。不過(guò)，你的自己定義好”反歸約”函數(shù)，即：該算子不僅有歸約函數(shù)（見(jiàn)參數(shù)func），還得有一個(gè)對(duì)應(yīng)的”反歸約”函數(shù)（見(jiàn)參數(shù)中的 invFunc）。和前面的reduceByKeyAndWindow() 類似，該算子也有一個(gè)可選參數(shù)numTasks來(lái)指定并行任務(wù)數(shù)。注意，這個(gè)算子需要配置好檢查點(diǎn)（checkpointing）才能用。

countByValueAndWindow(windowLength,slideInterval, [numTasks]) 基于包含(K, V)鍵值對(duì)的DStream，返回新的包含(K, Long)鍵值對(duì)的DStream。其中的Long value都是滑動(dòng)窗口內(nèi)key出現(xiàn)次數(shù)的計(jì)數(shù)。
和前面的reduceByKeyAndWindow() 類似，該算子也有一個(gè)可選參數(shù)numTasks來(lái)指定并行任務(wù)數(shù)。

最后，值得一提的是，你在Spark Streaming中做各種關(guān)聯(lián)（join）操作非常簡(jiǎn)單。

1、流-流（Stream-stream）關(guān)聯(lián)
一個(gè)數(shù)據(jù)流可以和另一個(gè)數(shù)據(jù)流直接關(guān)聯(lián)。

JavaPairDStream stream1 = ...
JavaPairDStream stream2 = ...
JavaPairDStream> joinedStream = stream1.join(stream2);

上面代碼中，stream1的每個(gè)批次中的RDD會(huì)和stream2相應(yīng)批次中的RDD進(jìn)行join。同樣，你可以類似地使用 leftOuterJoin, rightOuterJoin, fullOuterJoin 等。此外，你還可以基于窗口來(lái)join不同的數(shù)據(jù)流

JavaPairDStream windowedStream1 = stream1.window(Durations.seconds(20));
JavaPairDStream windowedStream2 = stream2.window(Durations.minutes(1));
JavaPairDStream> joinedStream = windowedStream1.join(windowedStream2);

2、流-數(shù)據(jù)集（stream-dataset）關(guān)聯(lián)
這里舉個(gè)基于滑動(dòng)窗口的例子。

JavaPairRDD dataset = ...
JavaPairDStream windowedStream = stream.window(Durations.seconds(20));
JavaPairDStream joinedStream = windowedStream.transform(
  new Function>, JavaRDD>>() {
    @Override
    public JavaRDD> call(JavaRDD> rdd) {
      return rdd.join(dataset);
    }
  }
);

在上面代碼里，你可以動(dòng)態(tài)地該表join的數(shù)據(jù)集（dataset）。傳給tranform算子的操作函數(shù)會(huì)在每個(gè)批次重新求值，所以每次該函數(shù)都會(huì)用最新的dataset值，所以不同批次間你可以改變dataset的值。

輸出算子可以將DStream的數(shù)據(jù)推送到外部系統(tǒng)，如：數(shù)據(jù)庫(kù)或者文件系統(tǒng)。因?yàn)檩敵鏊阕訒?huì)將最終完成轉(zhuǎn)換的數(shù)據(jù)輸出到外部系統(tǒng)，因此只有輸出算子調(diào)用時(shí)，才會(huì)真正觸發(fā)DStream transformation算子的真正執(zhí)行（這一點(diǎn)類似于RDD 的action算子）。目前所支持的輸出算子如下表：

print() 在驅(qū)動(dòng)器（driver）節(jié)點(diǎn)上打印DStream每個(gè)批次中的頭十個(gè)元素。

saveAsTextFiles(prefix, [suffix]) 將DStream的內(nèi)容保存到文本文件。
每個(gè)批次一個(gè)文件，各文件命名規(guī)則為 “prefix-TIME_IN_MS[.suffix]”

saveAsObjectFiles(prefix, [suffix]) 將DStream內(nèi)容以序列化Java對(duì)象的形式保存到順序文件中。
每個(gè)批次一個(gè)文件，各文件命名規(guī)則為 “prefix-TIME_IN_MS[.suffix]”Python API 暫不支持Python

saveAsHadoopFiles(prefix, [suffix]) 將DStream內(nèi)容保存到Hadoop文件中。
每個(gè)批次一個(gè)文件，各文件命名規(guī)則為 “prefix-TIME_IN_MS[.suffix]”Python API 暫不支持Python

foreachRDD(func) 這是最通用的輸出算子了，該算子接收一個(gè)函數(shù)func，func將作用于DStream的每個(gè)RDD上。
func應(yīng)該實(shí)現(xiàn)將每個(gè)RDD的數(shù)據(jù)推到外部系統(tǒng)中，比如：保存到文件或者寫到數(shù)據(jù)庫(kù)中。

注意，func函數(shù)是在streaming應(yīng)用的驅(qū)動(dòng)器進(jìn)程中執(zhí)行的，所以如果其中包含RDD的action算子，就會(huì)觸發(fā)對(duì)DStream中RDDs的實(shí)際計(jì)算過(guò)程。

DStream.foreachRDD是一個(gè)非常強(qiáng)大的原生工具函數(shù)，用戶可以基于此算子將DStream數(shù)據(jù)推送到外部系統(tǒng)中。不過(guò)用戶需要了解如何正確而高效地使用這個(gè)工具。以下列舉了一些常見(jiàn)的錯(cuò)誤。

通常，對(duì)外部系統(tǒng)寫入數(shù)據(jù)需要一些連接對(duì)象（如：遠(yuǎn)程server的TCP連接），以便發(fā)送數(shù)據(jù)給遠(yuǎn)程系統(tǒng)。因此，開(kāi)發(fā)人員可能會(huì)不經(jīng)意地在Spark驅(qū)動(dòng)器（driver）進(jìn)程中創(chuàng)建一個(gè)連接對(duì)象，然后又試圖在Spark worker節(jié)點(diǎn)上使用這個(gè)連接。如下例所示：

dstream.foreachRDD(new VoidFunction>() {
  @Override
  public void call(JavaRDD rdd) {
    final Connection connection = createNewConnection(); // executed at the driver
    rdd.foreach(new VoidFunction() {
      @Override
      public void call(String record) {
        connection.send(record); // executed at the worker
      }
    });
  }
});

這段代碼是錯(cuò)誤的，因?yàn)樗枰堰B接對(duì)象序列化，再?gòu)尿?qū)動(dòng)器節(jié)點(diǎn)發(fā)送到worker節(jié)點(diǎn)。而這些連接對(duì)象通常都是不能跨節(jié)點(diǎn)（機(jī)器）傳遞的。比如，連接對(duì)象通常都不能序列化，或者在另一個(gè)進(jìn)程中反序列化后再次初始化（連接對(duì)象通常都需要初始化，因此從驅(qū)動(dòng)節(jié)點(diǎn)發(fā)到worker節(jié)點(diǎn)后可能需要重新初始化）等。解決此類錯(cuò)誤的辦法就是在worker節(jié)點(diǎn)上創(chuàng)建連接對(duì)象。
一個(gè)比較好的解決方案是使用 rdd.foreachPartition – 為RDD的每個(gè)分區(qū)創(chuàng)建一個(gè)多帶帶的連接對(duì)象，示例如下：

dstream.foreachRDD(new VoidFunction>() {
  @Override
  public void call(JavaRDD rdd) {
    rdd.foreachPartition(new VoidFunction>() {
      @Override
      public void call(Iterator partitionOfRecords) {
        Connection connection = createNewConnection();
        while (partitionOfRecords.hasNext()) {
          connection.send(partitionOfRecords.next());
        }
        connection.close();
      }
    });
  }
});

最后，還有一個(gè)更優(yōu)化的辦法，就是在多個(gè)RDD批次之間復(fù)用連接對(duì)象。開(kāi)發(fā)者可以維護(hù)一個(gè)靜態(tài)連接池來(lái)保存連接對(duì)象，以便在不同批次的多個(gè)RDD之間共享同一組連接對(duì)象

dstream.foreachRDD(new VoidFunction>() {
  @Override
  public void call(JavaRDD rdd) {
    rdd.foreachPartition(new VoidFunction>() {
      @Override
      public void call(Iterator partitionOfRecords) {
        // ConnectionPool is a static, lazily initialized pool of connections
        Connection connection = ConnectionPool.getConnection();
        while (partitionOfRecords.hasNext()) {
          connection.send(partitionOfRecords.next());
        }
        ConnectionPool.returnConnection(connection); // return to the pool for future reuse
      }
    });
  }
});

注意，連接池中的連接應(yīng)該是懶惰創(chuàng)建的，并且有確定的超時(shí)時(shí)間，超時(shí)后自動(dòng)銷毀。這個(gè)實(shí)現(xiàn)應(yīng)該是目前發(fā)送數(shù)據(jù)最高效的實(shí)現(xiàn)方式。

注意點(diǎn)：

DStream的轉(zhuǎn)化執(zhí)行也是懶惰的，需要輸出算子來(lái)觸發(fā)，這一點(diǎn)和RDD的懶惰執(zhí)行由action算子觸發(fā)很類似。特別地，DStream輸出算子中包含的RDD action算子會(huì)強(qiáng)制觸發(fā)對(duì)所接收數(shù)據(jù)的處理。因此，如果你的Streaming應(yīng)用中沒(méi)有輸出算子，或者你用了dstream.foreachRDD(func)卻沒(méi)有在func中調(diào)用RDD action算子，那么這個(gè)應(yīng)用只會(huì)接收數(shù)據(jù)，而不會(huì)處理數(shù)據(jù)，接收到的數(shù)據(jù)最后只是被簡(jiǎn)單地丟棄掉了。

默認(rèn)地，輸出算子只能一次執(zhí)行一個(gè)，且按照它們?cè)趹?yīng)用程序代碼中定義的順序執(zhí)行。

首先需要注意的是，累加器（Accumulators）和廣播變量（Broadcast variables）是無(wú)法從Spark Streaming的檢查點(diǎn)中恢復(fù)回來(lái)的。所以如果你開(kāi)啟了檢查點(diǎn)功能，并同時(shí)在使用累加器和廣播變量，那么你最好是使用懶惰實(shí)例化的單例模式，因?yàn)檫@樣累加器和廣播變量才能在驅(qū)動(dòng)器（driver）故障恢復(fù)后重新實(shí)例化。

在Streaming應(yīng)用中可以調(diào)用DataFrames and SQL來(lái)處理流式數(shù)據(jù)。開(kāi)發(fā)者可以用通過(guò)StreamingContext中的SparkContext對(duì)象來(lái)創(chuàng)建一個(gè)SQLContext，并且，開(kāi)發(fā)者需要確保一旦驅(qū)動(dòng)器（driver）故障恢復(fù)后，該SQLContext對(duì)象能重新創(chuàng)建出來(lái)。同樣，你還是可以使用懶惰創(chuàng)建的單例模式來(lái)實(shí)例化SQLContext，如下面的代碼所示，這里我們將最開(kāi)始的那個(gè)小栗子做了一些修改，使用DataFrame和SQL來(lái)統(tǒng)計(jì)單詞計(jì)數(shù)。其實(shí)就是，將每個(gè)RDD都轉(zhuǎn)化成一個(gè)DataFrame，然后注冊(cè)成臨時(shí)表，再用SQL查詢這些臨時(shí)表。

與RDD類似，Spark Streaming也可以讓開(kāi)發(fā)人員手動(dòng)控制，將數(shù)據(jù)流中的數(shù)據(jù)持久化到內(nèi)存中。對(duì)DStream調(diào)用persist()方法，就可以讓Spark Streaming自動(dòng)將該數(shù)據(jù)流中的所有產(chǎn)生的RDD，都持久化到內(nèi)存中。如果要對(duì)一個(gè)DStream多次執(zhí)行操作，那么，對(duì)DStream持久化是非常有用的。因?yàn)槎啻尾僮鳎梢怨蚕硎褂脙?nèi)存中的一份緩存數(shù)據(jù)。

對(duì)于基于窗口的操作，比如reduceByWindow、reduceByKeyAndWindow，以及基于狀態(tài)的操作，比如updateStateByKey，默認(rèn)就隱式開(kāi)啟了持久化機(jī)制。即Spark Streaming默認(rèn)就會(huì)將上述操作產(chǎn)生的Dstream中的數(shù)據(jù)，緩存到內(nèi)存中，不需要開(kāi)發(fā)人員手動(dòng)調(diào)用persist()方法。

對(duì)于通過(guò)網(wǎng)絡(luò)接收數(shù)據(jù)的輸入流，比如socket、Kafka、Flume等，默認(rèn)的持久化級(jí)別，是將數(shù)據(jù)復(fù)制一份，以便于容錯(cuò)。相當(dāng)于是，用的是類似MEMORY_ONLY_SER_2。

與RDD不同的是，默認(rèn)的持久化級(jí)別，統(tǒng)一都是要序列化的。

在Spark web UI上看到多出了一個(gè)Streaming tab頁(yè)，上面顯示了正在運(yùn)行的接收器（是否活躍，接收記錄的條數(shù)，失敗信息等）和處理完的批次信息（批次處理時(shí)間，查詢延時(shí)等）。這些信息都可以用來(lái)監(jiān)控streaming應(yīng)用。
web UI上有兩個(gè)度量特別重要：

批次處理耗時(shí)（Processing Time） – 處理單個(gè)批次耗時(shí)

批次調(diào)度延時(shí)（Scheduling Delay） -各批次在隊(duì)列中等待時(shí)間（等待上一個(gè)批次處理完）

如果批次處理耗時(shí)一直比批次間隔時(shí)間大，或者批次調(diào)度延時(shí)持續(xù)上升，就意味著系統(tǒng)處理速度跟不上數(shù)據(jù)接收速度。這時(shí)候你就得考慮一下怎么把批次處理時(shí)間降下來(lái)（reducing）。

Spark Streaming程序的處理進(jìn)度可以用StreamingListener接口來(lái)監(jiān)聽(tīng)，這個(gè)接口可以監(jiān)聽(tīng)到接收器的狀態(tài)和處理時(shí)間。

要想streaming應(yīng)用在集群上穩(wěn)定運(yùn)行，那么系統(tǒng)處理數(shù)據(jù)的速度必須能跟上其接收數(shù)據(jù)的速度。換句話說(shuō)，批次數(shù)據(jù)的處理速度應(yīng)該和其生成速度一樣快。對(duì)于特定的應(yīng)用來(lái)說(shuō)，可以從其對(duì)應(yīng)的監(jiān)控（monitoring）頁(yè)面上觀察驗(yàn)證，頁(yè)面上顯示的處理耗時(shí)應(yīng)該要小于批次間隔時(shí)間。

根據(jù)spark streaming計(jì)算的性質(zhì)，在一定的集群資源限制下，批次間隔的值會(huì)極大地影響系統(tǒng)的數(shù)據(jù)處理能力。例如，在WordCountNetwork示例中，對(duì)于特定的數(shù)據(jù)速率，一個(gè)系統(tǒng)可能能夠在批次間隔為2秒時(shí)跟上數(shù)據(jù)接收速度，但如果把批次間隔改為500毫秒系統(tǒng)可能就處理不過(guò)來(lái)了。所以，批次間隔需要謹(jǐn)慎設(shè)置，以確保生產(chǎn)系統(tǒng)能夠處理得過(guò)來(lái)。

要找出適合的批次間隔，你可以從一個(gè)比較保守的批次間隔值（如5~10秒）開(kāi)始測(cè)試。要驗(yàn)證系統(tǒng)是否能跟上當(dāng)前的數(shù)據(jù)接收速率，你可能需要檢查一下端到端的批次處理延遲（可以看看Spark驅(qū)動(dòng)器log4j日志中的Total delay，也可以用StreamingListener接口來(lái)檢測(cè)）。如果這個(gè)延遲能保持和批次間隔差不多，那么系統(tǒng)基本就是穩(wěn)定的。否則，如果這個(gè)延遲持久在增長(zhǎng)，也就是說(shuō)系統(tǒng)跟不上數(shù)據(jù)接收速度，那也就意味著系統(tǒng)不穩(wěn)定。一旦系統(tǒng)文檔下來(lái)后，你就可以嘗試提高數(shù)據(jù)接收速度，或者減少批次間隔值。不過(guò)需要注意，瞬間的延遲增長(zhǎng)可以只是暫時(shí)的，只要這個(gè)延遲后續(xù)會(huì)自動(dòng)降下來(lái)就沒(méi)有問(wèn)題（如：降到小于批次間隔值）

http://ifeve.com/spark-stream...
http://spark.apache.org/docs/...