摘要:輸入和接收器輸入代表從某種流式數據源流入的數據流。文件數據流可以從任何兼容包括等的文件系統,創建方式如下將監視該目錄,并處理該目錄下任何新建的文件目前還不支持嵌套目錄。會被一個個依次推入隊列,而則會依次以數據流形式處理這些的數據。
特點:
Spark Streaming能夠實現對實時數據流的流式處理,并具有很好的可擴展性、高吞吐量和容錯性。
Spark Streaming支持從多種數據源提取數據,如:Kafka、Flume、Twitter、ZeroMQ、Kinesis以及TCP套接字,并且可以提供一些高級API來表達復雜的處理算法,如:map、reduce、join和window等。
Spark Streaming支持將處理完的數據推送到文件系統、數據庫或者實時儀表盤中展示。
可以將Spark的機器學習(machine learning) 和 圖計算(graph processing)的算法應用于Spark Streaming的數據流當中。
下圖展示了Spark Streaming的內部工作原理。Spark Streaming從實時數據流接入數據,再將其劃分為一個個小批量供后續Spark engine處理,所以實際上,Spark Streaming是按一個個小批量來處理數據流的。
Spark Streaming為這種持續的數據流提供了的一個高級抽象,即:discretized stream(離散數據流)或者叫DStream。DStream既可以從輸入數據源創建得來,如:Kafka、Flume或者Kinesis,也可以從其他DStream經一些算子操作得到。其實在內部,一個DStream就是包含了一系列RDDs。
入門實例分析SparkConf conf = new SparkConf().setAppName("stream1").setMaster("local[2]"); JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(1)); JavaReceiverInputDStreamlines = 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的入口。并將批次間隔設為1秒。
利用這個上下文對象(StreamingContext),我們可以創建一個DStream,該DStream代表從前面的TCP數據源流入的數據流,同時TCP數據源是由主機名(如:hostnam)和端口(如:9999)來描述的。
這里的 lines 就是從數據server接收到的數據流。其中每一條記錄都是一行文本。接下來,我們就需要把這些文本行按空格分割成單詞。
flatMap 是一種 “一到多”(one-to-many)的映射算子,它可以將源DStream中每一條記錄映射成多條記錄,從而產生一個新的DStream對象。在本例中,lines中的每一行都會被flatMap映射為多個單詞,從而生成新的words DStream對象。然后,我們就能對這些單詞進行計數了。
words這個DStream對象經過map算子(一到一的映射)轉換為一個包含(word, 1)鍵值對的DStream對象pairs,再對pairs使用reduce算子,得到每個批次中各個單詞的出現頻率。
注意,執行以上代碼后,Spark Streaming只是將計算邏輯設置好,此時并未真正的開始處理數據。要啟動之前的處理邏輯,我們還需要如下調用:
ssc.start() // 啟動流式計算 ssc.awaitTermination() // 等待直到計算終止
首先,你需要運行netcat(Unix-like系統都會有這個小工具),將其作為data server
$ nc -lk 9999
然后,執行程序. 現在你嘗試可以在運行netcat的終端里敲幾個單詞,你會發現這些單詞以及相應的計數會出現在啟動Spark Streaming例子的終端屏幕上。
注意,StreamingContext在內部會創建一個 SparkContext 對象(SparkContext是所有Spark應用的入口,在StreamingContext對象中可以這樣訪問:ssc.sparkContext)。
StreamingContext還有另一個構造參數,即:批次間隔,這個值的大小需要根據應用的具體需求和可用的集群資源來確定。
需要關注的重點:
一旦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都包含了特定時間間隔內的一批數據,如下圖所示:
任何作用于DStream的算子,其實都會被轉化為對其內部RDD的操作。底層的RDD轉換仍然是由Spark引擎來計算。DStream的算子將這些細節隱藏了起來,并為開發者提供了更為方便的高級API。
輸入DStream和接收器輸入DStream代表從某種流式數據源流入的數據流。在之前的例子里,lines 對象就是輸入DStream,它代表從netcat server收到的數據流。每個輸入DStream(除文件數據流外)都和一個接收器(Receiver)相關聯,而接收器則是專門從數據源拉取數據到內存中的對象。
Spark Streaming主要提供兩種內建的流式數據源:
基礎數據源(Basic sources): 在StreamingContext API 中可直接使用的源,如:文件系統,套接字連接或者Akka actor。
高級數據源(Advanced sources): 需要依賴額外工具類的源,如:Kafka、Flume、Kinesis、Twitter等數據源。這些數據源都需要增加額外的依賴,詳見依賴鏈接(linking)這一節。
注意,如果你需要同時從多個數據源拉取數據,那么你就需要創建多個DStream對象。多個DStream對象其實也就同時創建了多個數據流接收器。但是請注意,Spark的worker/executor 都是長期運行的,因此它們都會各自占用一個分配給Spark Streaming應用的CPU。
因此,本地運行時,一定要將master設為”local[n]”,其中 n > 接收器的個數。
將Spark Streaming應用置于集群中運行時,同樣,分配給該應用的CPU core數必須大于接收器的總數。否則,該應用就只會接收數據,而不會處理數據。
使用ssc.socketTextStream(…) 可以從一個TCP連接中接收文本數據。而除了TCP套接字外,StreamingContext API 還支持從文件或者Akka actor中拉取數據。
文件數據流(File Streams): 可以從任何兼容HDFS API(包括:HDFS、S3、NFS等)的文件系統,創建方式如下:
streamingContext.fileStream(dataDirectory);
Spark Streaming將監視該dataDirectory目錄,并處理該目錄下任何新建的文件(目前還不支持嵌套目錄)。注意:
各個文件數據格式必須一致。
dataDirectory中的文件必須通過moving或者renaming來創建。
一旦文件move進dataDirectory之后,就不能再改動。所以如果這個文件后續還有寫入,這些新寫入的數據不會被讀取。
對于簡單的文本文件,更簡單的方式是調用 streamingContext.textFileStream(dataDirectory)。
另外,文件數據流不是基于接收器的,所以不需要為其多帶帶分配一個CPU core。
RDD隊列數據流(Queue of RDDs as a Stream): 如果需要測試Spark Streaming應用,你可以創建一個基于一批RDD的DStream對象,只需調用 streamingContext.queueStream(queueOfRDDs)。RDD會被一個個依次推入隊列,而DStream則會依次以數據流形式處理這些RDD的數據。
自定義數據源
輸入DStream也可以用自定義的方式創建。你需要做的只是實現一個自定義的接收器(receiver),以便從自定義的數據源接收數據,然后將數據推入Spark中。 見:http://spark.apache.org/docs/...
接收器可靠性
從可靠性角度來劃分,大致有兩種數據源。其中,像Kafka、Flume這樣的數據源,它們支持對所傳輸的數據進行確認。系統收到這類可靠數據源過來的數據,然后發出確認信息,這樣就能夠確保任何失敗情況下,都不會丟數據。因此我們可以將接收器也相應地分為兩類:
可靠接收器(Reliable Receiver) – 可靠接收器會在成功接收并保存好Spark數據副本后,向可靠數據源發送確認信息。
不可靠接收器(Unreliable Receiver) – 不可靠接收器不會發送任何確認信息。
DStream支持的transformation算子和RDD類似,DStream也支持從輸入DStream經過各種transformation算子映射成新的DStream。
map(func) 返回會一個新的DStream,并將源DStream中每個元素通過func映射為新的元素
flatMap(func) 和map類似,不過每個輸入元素不再是映射為一個輸出,而是映射為0到多個輸出
filter(func) 返回一個新的DStream,并包含源DStream中被func選中(func返回true)的元素
repartition(numPartitions) 更改DStream的并行度(增加或減少分區數)
union(otherStream) 返回新的DStream,包含源DStream和otherDStream元素的并集
count() 返回一個包含單元素RDDs的DStream,其中每個元素是源DStream中各個RDD中的元素個數
reduce(func) 返回一個包含單元素RDDs的DStream,其中每個元素是通過源RDD中各個RDD的元素經func(func輸入兩個參數并返回一個同類型結果數據)聚合得到的結果。func必須滿足結合律,以便支持并行計算。
countByValue() 如果源DStream包含的元素類型為K,那么該算子返回新的DStream包含元素為(K, Long)鍵值對,其中K為源DStream各個元素,而Long為該元素出現的次數。
reduceByKey(func, [numTasks]) 如果源DStream 包含的元素為 (K, V) 鍵值對,則該算子返回一個新的也包含(K, V)鍵值對的DStream,其中V是由func聚合得到的。注意:默認情況下,該算子使用Spark的默認并發任務數(本地模式為2,集群模式下由spark.default.parallelism 決定)。你可以通過可選參數numTasks來指定并發任務個數。
join(otherStream, [numTasks]) 如果源DStream包含元素為(K, V),同時otherDStream包含元素為(K, W)鍵值對,則該算子返回一個新的DStream,其中源DStream和otherDStream中每個K都對應一個 (K, (V, W))鍵值對元素。
cogroup(otherStream, [numTasks]) 如果源DStream包含元素為(K, V),同時otherDStream包含元素為(K, W)鍵值對,則該算子返回一個新的DStream,其中每個元素類型為包含(K, Seq[V], Seq[W])的tuple。
transform(func) 返回一個新的DStream,其包含的RDD為源RDD經過func操作后得到的結果。利用該算子可以對DStream施加任意的操作。
updateStateByKey(func) 返回一個包含新”狀態”的DStream。源DStream中每個key及其對應的values會作為func的輸入,而func可以用于對每個key的“狀態”數據作任意的更新操作。
updateStateByKey算子updateStateByKey 算子支持維護一個任意的狀態。要實現這一點,只需要兩步:
定義狀態 – 狀態數據可以是任意類型。
定義狀態更新函數 – 定義好一個函數,其輸入為數據流之前的狀態和新的數據流數據,且可其更新步驟1中定義的輸入數據流的狀態。
在每一個批次數據到達后,Spark都會調用狀態更新函數,來更新所有已有key(不管key是否存在于本批次中)的狀態。如果狀態更新函數返回None,則對應的鍵值對會被刪除。
舉例如下。假設你需要維護一個流式應用,統計數據流中每個單詞的出現次數。這里將各個單詞的出現次數這個整型數定義為狀態。我們接下來定義狀態更新函數如下:
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); } };
注意,調用updateStateByKey前需要配置檢查點目錄. 配置方式見下:
檢查點一般來說Streaming 應用都需要7*24小時長期運行,所以必須對一些與業務邏輯無關的故障有很好的容錯(如:系統故障、JVM崩潰等)。對于這些可能性,Spark Streaming 必須在檢查點保存足夠的信息到一些可容錯的外部存儲系統中,以便能夠隨時從故障中恢復回來。所以,檢查點需要保存以下兩種數據:
元數據檢查點(Metadata checkpointing) – 保存流式計算邏輯的定義信息到外部可容錯存儲系統(如:HDFS)。主要用途是用于在故障后回復應用程序本身(后續詳談)。元數包括:
Configuration – 創建Streaming應用程序的配置信息。
DStream operations – 定義流式處理邏輯的DStream操作信息。
Incomplete batches – 已經排隊但未處理完的批次信息。
總之,元數據檢查點主要是為了恢復驅動器節點上的故障,而數據或RDD檢查點是為了支持對有狀態轉換操作的恢復。
何時啟用檢查點
使用了有狀態的轉換算子(Usage of stateful transformations) – 不管是用了 updateStateByKey 還是用了 reduceByKeyAndWindow(有”反歸約”函數的那個版本),你都必須配置檢查點目錄來周期性地保存RDD檢查點。
支持驅動器故障中恢復(Recovering from failures of the driver running the application) – 這時候需要元數據檢查點以便恢復流式處理的進度信息。
注意,一些簡單的流式應用,如果沒有用到前面所說的有狀態轉換算子,則完全可以不開啟檢查點。不過這樣的話,驅動器(driver)故障恢復后,有可能會丟失部分數據(有些已經接收但還未處理的數據可能會丟失)。不過通常這點丟失時可接受的,很多Spark Streaming應用也是這樣運行的。
如何配置檢查點
檢查點的啟用,只需要設置好保存檢查點信息的檢查點目錄即可,一般會會將這個目錄設為一些可容錯的、可靠性較高的文件系統(如:HDFS、S3等)。
第一種:開發者只需要調用 streamingContext.checkpoint(checkpointDirectory)。設置好檢查點,你就可以使用前面提到的有狀態轉換算子了。
第二種:如果你需要你的應用能夠支持從驅動器故障中恢復,你可能需要重寫部分代碼,實現以下行為:
如果程序是首次啟動,就需要new一個新的StreamingContext,并定義好所有的數據流處理,然后調用StreamingContext.start()。
如果程序是故障后重啟,就需要從檢查點目錄中的數據中重新構建StreamingContext對象。
// 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 JavaDStreamlines = 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檢查點會增加額外的保存數據的開銷。這可能會導致數據流的處理時間變長。
因此,你必須仔細的調整檢查點間隔時間。如果批次間隔太小(比如:1秒),那么對每個批次保存檢查點數據將大大減小吞吐量。
另一方面,檢查點保存過于頻繁又會導致血統信息和任務個數的增加,這同樣會影響系統性能。
對于需要RDD檢查點的有狀態轉換算子,默認的間隔是批次間隔的整數倍,且最小10秒。開發人員可以這樣來自定義這個間隔:dstream.checkpoint(checkpointInterval)。一般推薦設為批次間隔時間的5~10倍。
transform算子transform算子(及其變體transformWith)可以支持任意的RDD到RDD的映射操作。也就是說,你可以用tranform算子來包裝任何DStream API所不支持的RDD算子。例如,將DStream每個批次中的RDD和另一個Dataset進行關聯(join)操作,這個功能DStream API并沒有直接支持。不過你可以用transform來實現這個功能,可見transform其實為DStream提供了非常強大的功能支持。比如說,你可以用事先算好的垃圾信息,對DStream進行實時過濾。
// RDD containing spam information final JavaPairRDDspamInfoRDD = 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包含的算子,其調用時間間隔和批次間隔是相同的。所以你可以基于時間改變對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搞定。
// Reduce function adding two integers, defined separately for clarity Function2reduceFunc = new Function2 () { @Override public Integer call(Integer i1, Integer i2) { return i1 + i2; } }; // 每隔10秒歸約一次最近30秒的數據 JavaPairDStream windowedWordCounts = pairs.reduceByKeyAndWindow(reduceFunc, Durations.seconds(30), Durations.seconds(10));
以下列出了常用的窗口算子。所有這些算子都有前面提到的那兩個參數 – 窗口長度 和 滑動距離。
window(windowLength, slideInterval) 將源DStream窗口化,并返回轉化后的DStream
countByWindow(windowLength,slideInterval) 返回數據流在一個滑動窗口內的元素個數
reduceByWindow(func, windowLength,slideInterval) 基于數據流在一個滑動窗口內的元素,用func做聚合,返回一個單元素數據流。func必須滿足結合律,以便支持并行計算。
reduceByKeyAndWindow(func,windowLength, slideInterval, [numTasks]) 基于(K, V)鍵值對DStream,將一個滑動窗口內的數據進行聚合,返回一個新的包含(K,V)鍵值對的DStream,其中每個value都是各個key經過func聚合后的結果。
注意:如果不指定numTasks,其值將使用Spark的默認并行任務數(本地模式下為2,集群模式下由 spark.default.parallelism決定)。當然,你也可以通過numTasks來指定任務個數。
reduceByKeyAndWindow(func, invFunc,windowLength,slideInterval, [numTasks]) 和前面的reduceByKeyAndWindow() 類似,只是這個版本會用之前滑動窗口計算結果,遞增地計算每個窗口的歸約結果。當新的數據進入窗口時,這些values會被輸入func做歸約計算,而這些數據離開窗口時,對應的這些values又會被輸入 invFunc 做”反歸約”計算。舉個簡單的例子,就是把新進入窗口數據中各個單詞個數“增加”到各個單詞統計結果上,同時把離開窗口數據中各個單詞的統計個數從相應的統計結果中“減掉”。不過,你的自己定義好”反歸約”函數,即:該算子不僅有歸約函數(見參數func),還得有一個對應的”反歸約”函數(見參數中的 invFunc)。和前面的reduceByKeyAndWindow() 類似,該算子也有一個可選參數numTasks來指定并行任務數。注意,這個算子需要配置好檢查點(checkpointing)才能用。
countByValueAndWindow(windowLength,slideInterval, [numTasks]) 基于包含(K, V)鍵值對的DStream,返回新的包含(K, Long)鍵值對的DStream。其中的Long value都是滑動窗口內key出現次數的計數。
和前面的reduceByKeyAndWindow() 類似,該算子也有一個可選參數numTasks來指定并行任務數。
最后,值得一提的是,你在Spark Streaming中做各種關聯(join)操作非常簡單。
1、流-流(Stream-stream)關聯
一個數據流可以和另一個數據流直接關聯。
JavaPairDStreamstream1 = ... JavaPairDStream stream2 = ... JavaPairDStream > joinedStream = stream1.join(stream2);
上面代碼中,stream1的每個批次中的RDD會和stream2相應批次中的RDD進行join。同樣,你可以類似地使用 leftOuterJoin, rightOuterJoin, fullOuterJoin 等。此外,你還可以基于窗口來join不同的數據流
JavaPairDStreamwindowedStream1 = stream1.window(Durations.seconds(20)); JavaPairDStream windowedStream2 = stream2.window(Durations.minutes(1)); JavaPairDStream > joinedStream = windowedStream1.join(windowedStream2);
2、流-數據集(stream-dataset)關聯
這里舉個基于滑動窗口的例子。
JavaPairRDDdataset = ... JavaPairDStream windowedStream = stream.window(Durations.seconds(20)); JavaPairDStream joinedStream = windowedStream.transform( new Function >, JavaRDD >>() { @Override public JavaRDD > call(JavaRDD > rdd) { return rdd.join(dataset); } } );
在上面代碼里,你可以動態地該表join的數據集(dataset)。傳給tranform算子的操作函數會在每個批次重新求值,所以每次該函數都會用最新的dataset值,所以不同批次間你可以改變dataset的值。
DStream輸出算子輸出算子可以將DStream的數據推送到外部系統,如:數據庫或者文件系統。因為輸出算子會將最終完成轉換的數據輸出到外部系統,因此只有輸出算子調用時,才會真正觸發DStream transformation算子的真正執行(這一點類似于RDD 的action算子)。目前所支持的輸出算子如下表:
print() 在驅動器(driver)節點上打印DStream每個批次中的頭十個元素。
saveAsTextFiles(prefix, [suffix]) 將DStream的內容保存到文本文件。
每個批次一個文件,各文件命名規則為 “prefix-TIME_IN_MS[.suffix]”
saveAsObjectFiles(prefix, [suffix]) 將DStream內容以序列化Java對象的形式保存到順序文件中。
每個批次一個文件,各文件命名規則為 “prefix-TIME_IN_MS[.suffix]”Python API 暫不支持Python
saveAsHadoopFiles(prefix, [suffix]) 將DStream內容保存到Hadoop文件中。
每個批次一個文件,各文件命名規則為 “prefix-TIME_IN_MS[.suffix]”Python API 暫不支持Python
foreachRDD(func) 這是最通用的輸出算子了,該算子接收一個函數func,func將作用于DStream的每個RDD上。
func應該實現將每個RDD的數據推到外部系統中,比如:保存到文件或者寫到數據庫中。
注意,func函數是在streaming應用的驅動器進程中執行的,所以如果其中包含RDD的action算子,就會觸發對DStream中RDDs的實際計算過程。
使用foreachRDD的設計模式DStream.foreachRDD是一個非常強大的原生工具函數,用戶可以基于此算子將DStream數據推送到外部系統中。不過用戶需要了解如何正確而高效地使用這個工具。以下列舉了一些常見的錯誤。
通常,對外部系統寫入數據需要一些連接對象(如:遠程server的TCP連接),以便發送數據給遠程系統。因此,開發人員可能會不經意地在Spark驅動器(driver)進程中創建一個連接對象,然后又試圖在Spark worker節點上使用這個連接。如下例所示:
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 } }); } });
這段代碼是錯誤的,因為它需要把連接對象序列化,再從驅動器節點發送到worker節點。而這些連接對象通常都是不能跨節點(機器)傳遞的。比如,連接對象通常都不能序列化,或者在另一個進程中反序列化后再次初始化(連接對象通常都需要初始化,因此從驅動節點發到worker節點后可能需要重新初始化)等。解決此類錯誤的辦法就是在worker節點上創建連接對象。
一個比較好的解決方案是使用 rdd.foreachPartition – 為RDD的每個分區創建一個多帶帶的連接對象,示例如下:
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(); } }); } });
最后,還有一個更優化的辦法,就是在多個RDD批次之間復用連接對象。開發者可以維護一個靜態連接池來保存連接對象,以便在不同批次的多個RDD之間共享同一組連接對象
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 } }); } });
注意,連接池中的連接應該是懶惰創建的,并且有確定的超時時間,超時后自動銷毀。這個實現應該是目前發送數據最高效的實現方式。
注意點:
DStream的轉化執行也是懶惰的,需要輸出算子來觸發,這一點和RDD的懶惰執行由action算子觸發很類似。特別地,DStream輸出算子中包含的RDD action算子會強制觸發對所接收數據的處理。因此,如果你的Streaming應用中沒有輸出算子,或者你用了dstream.foreachRDD(func)卻沒有在func中調用RDD action算子,那么這個應用只會接收數據,而不會處理數據,接收到的數據最后只是被簡單地丟棄掉了。
默認地,輸出算子只能一次執行一個,且按照它們在應用程序代碼中定義的順序執行。
累加器和廣播變量首先需要注意的是,累加器(Accumulators)和廣播變量(Broadcast variables)是無法從Spark Streaming的檢查點中恢復回來的。所以如果你開啟了檢查點功能,并同時在使用累加器和廣播變量,那么你最好是使用懶惰實例化的單例模式,因為這樣累加器和廣播變量才能在驅動器(driver)故障恢復后重新實例化。
DataFrame和SQL相關算子在Streaming應用中可以調用DataFrames and SQL來處理流式數據。開發者可以用通過StreamingContext中的SparkContext對象來創建一個SQLContext,并且,開發者需要確保一旦驅動器(driver)故障恢復后,該SQLContext對象能重新創建出來。同樣,你還是可以使用懶惰創建的單例模式來實例化SQLContext,如下面的代碼所示,這里我們將最開始的那個小栗子做了一些修改,使用DataFrame和SQL來統計單詞計數。其實就是,將每個RDD都轉化成一個DataFrame,然后注冊成臨時表,再用SQL查詢這些臨時表。
緩存與持久化機制與RDD類似,Spark Streaming也可以讓開發人員手動控制,將數據流中的數據持久化到內存中。對DStream調用persist()方法,就可以讓Spark Streaming自動將該數據流中的所有產生的RDD,都持久化到內存中。如果要對一個DStream多次執行操作,那么,對DStream持久化是非常有用的。因為多次操作,可以共享使用內存中的一份緩存數據。
對于基于窗口的操作,比如reduceByWindow、reduceByKeyAndWindow,以及基于狀態的操作,比如updateStateByKey,默認就隱式開啟了持久化機制。即Spark Streaming默認就會將上述操作產生的Dstream中的數據,緩存到內存中,不需要開發人員手動調用persist()方法。
對于通過網絡接收數據的輸入流,比如socket、Kafka、Flume等,默認的持久化級別,是將數據復制一份,以便于容錯。相當于是,用的是類似MEMORY_ONLY_SER_2。
與RDD不同的是,默認的持久化級別,統一都是要序列化的。
應用監控在Spark web UI上看到多出了一個Streaming tab頁,上面顯示了正在運行的接收器(是否活躍,接收記錄的條數,失敗信息等)和處理完的批次信息(批次處理時間,查詢延時等)。這些信息都可以用來監控streaming應用。
web UI上有兩個度量特別重要:
批次處理耗時(Processing Time) – 處理單個批次耗時
批次調度延時(Scheduling Delay) -各批次在隊列中等待時間(等待上一個批次處理完)
如果批次處理耗時一直比批次間隔時間大,或者批次調度延時持續上升,就意味著系統處理速度跟不上數據接收速度。這時候你就得考慮一下怎么把批次處理時間降下來(reducing)。
Spark Streaming程序的處理進度可以用StreamingListener接口來監聽,這個接口可以監聽到接收器的狀態和處理時間。
設置合適的批次間隔要想streaming應用在集群上穩定運行,那么系統處理數據的速度必須能跟上其接收數據的速度。換句話說,批次數據的處理速度應該和其生成速度一樣快。對于特定的應用來說,可以從其對應的監控(monitoring)頁面上觀察驗證,頁面上顯示的處理耗時應該要小于批次間隔時間。
根據spark streaming計算的性質,在一定的集群資源限制下,批次間隔的值會極大地影響系統的數據處理能力。例如,在WordCountNetwork示例中,對于特定的數據速率,一個系統可能能夠在批次間隔為2秒時跟上數據接收速度,但如果把批次間隔改為500毫秒系統可能就處理不過來了。所以,批次間隔需要謹慎設置,以確保生產系統能夠處理得過來。
要找出適合的批次間隔,你可以從一個比較保守的批次間隔值(如5~10秒)開始測試。要驗證系統是否能跟上當前的數據接收速率,你可能需要檢查一下端到端的批次處理延遲(可以看看Spark驅動器log4j日志中的Total delay,也可以用StreamingListener接口來檢測)。如果這個延遲能保持和批次間隔差不多,那么系統基本就是穩定的。否則,如果這個延遲持久在增長,也就是說系統跟不上數據接收速度,那也就意味著系統不穩定。一旦系統文檔下來后,你就可以嘗試提高數據接收速度,或者減少批次間隔值。不過需要注意,瞬間的延遲增長可以只是暫時的,只要這個延遲后續會自動降下來就沒有問題(如:降到小于批次間隔值)
參考:http://ifeve.com/spark-stream...
http://spark.apache.org/docs/...
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/67260.html
摘要:用來管理文件系統的命名空間,其將所有的文件和文件夾的元數據保存在一個文件系統樹中。文件系統鏡像元數據鏡像文件。其周期性的向元數據節點回報其存儲的數據塊信息。 Hadoop分布式文件系統(hadoopdistributed filesystem,HDFS)。HDFS是一個高度容錯性的系統,適合部署在廉價的機器上。HDFS能提供高吞吐量的數據訪問,非常適合大規模數據集上的應用。HDFS可以...
摘要:原文鏈接這些年,你不能錯過的學習資源寫在前面本系列是綜合了自己在學習過程中的理解記錄對參考文章中的一些理解個人實踐過程中的一些心得而來。 原文鏈接:『 Spark 』5. 這些年,你不能錯過的 spark 學習資源 寫在前面 本系列是綜合了自己在學習spark過程中的理解記錄 + 對參考文章中的一些理解 + 個人實踐spark過程中的一些心得而來。寫這樣一個系列僅僅是為了梳理個人學習s...
摘要:原文鏈接簡介寫在前面本系列是綜合了自己在學習過程中的理解記錄對參考文章中的一些理解個人實踐過程中的一些心得而來。其次,本系列是基于目前最新的系列開始的,目前的更新速度很快,記錄一下版本好還是必要的。 原文鏈接:『 Spark 』1. spark 簡介 寫在前面 本系列是綜合了自己在學習spark過程中的理解記錄 + 對參考文章中的一些理解 + 個人實踐spark過程中的一些心得而來。寫...
閱讀 3725·2021-09-22 10:57
閱讀 1914·2019-08-30 15:55
閱讀 2699·2019-08-30 15:44
閱讀 1731·2019-08-30 15:44
閱讀 1876·2019-08-30 15:44
閱讀 2244·2019-08-30 12:49
閱讀 1053·2019-08-29 18:47
閱讀 3135·2019-08-29 16:15