Spark Streaming 技術(shù)點(diǎn)匯總

Spark Streaming 支持實(shí)時(shí)數(shù)據(jù)流的可擴(kuò)展(Scalable)、高吞吐(high-throughput)、容錯(fault-tolerant)的流處理(stream processing)。
Spark Streaming 支持實(shí)時(shí)數(shù)據(jù)流的可擴(kuò)展(Scalable)、高吞吐(high-throughput)、容錯(fault-tolerant)的流處理(stream processing)。

架構(gòu)圖
特性如下：
? 可線性伸縮至超過數(shù)百個(gè)節(jié)點(diǎn);
? 實(shí)現(xiàn)亞秒級延遲處理;
? 可與 Spark 批處理和交互式處理無縫集成;
? 提供簡單的API實(shí)現(xiàn)復(fù)雜算法;
? 更多的流方式支持，包括 Kafka、Flume、Kinesis、Twitter、ZeroMQ 等。
001、原理
Spark 在接收到實(shí)時(shí)輸入數(shù)據(jù)流后，將數(shù)據(jù)劃分成批次(divides the data into batches)，然后轉(zhuǎn)給 Spark Engine 處理，按批次生成最后的結(jié)果流(generate the final stream of results in batches)。

002、API
DStream：
DStream(Discretized Stream，離散流)是 Spark Stream 提供的高級抽象連續(xù)數(shù)據(jù)流。
組成：一個(gè) DStream 可看作一個(gè) RDDs 序列。
核心思想：將計(jì)算作為一系列較小時(shí)間間隔的、狀態(tài)無關(guān)的、確定批次的任務(wù)，每個(gè)時(shí)間間隔內(nèi)接收的輸入數(shù)據(jù)被可靠存儲在集群中，作為一個(gè)輸入數(shù)據(jù)集。

特性：一個(gè)高層次的函數(shù)式編程 API、強(qiáng)一致性以及高校的故障恢復(fù)。
應(yīng)用程序模板：
模板1

模板2

WordCount示例

Input DStream：
Input DStream 是一種從流式數(shù)據(jù)源獲取原始數(shù)據(jù)流的 DStream，分為基本輸入源(文件系統(tǒng)、Socket、Akka Actor、自定義數(shù)據(jù)源)和高級輸入源(Kafka、Flume等)。
Receiver：
每個(gè) Input DStream(文件流除外)都會對應(yīng)一個(gè)單一的 Receiver對象，負(fù)責(zé)從數(shù)據(jù)源接收數(shù)據(jù)并存入 Spark 內(nèi)存進(jìn)行處理。應(yīng)用程序中可創(chuàng)建多個(gè) Input DStream 并行接收多個(gè)數(shù)據(jù)流。
每個(gè) Receiver 是一個(gè)長期運(yùn)行在Worker或者 Executor 上的 Task，所以會占用該應(yīng)用程序的一個(gè)核(core)。如果分配給 Spark Streaming 應(yīng)用程序的核數(shù)小于或等于 Input DStream 個(gè)數(shù)(即Receiver個(gè)數(shù))，則只能接收數(shù)據(jù)，卻沒有能力全部處理(文件流除外，因?yàn)闊o需Receiver)。
Spark Streaming 已封裝各種數(shù)據(jù)源，需要時(shí)參考官方文檔。
Transformation Operation
常用Transformation

updateStateByKey(func)
updateStateByKey可對DStream中的數(shù)據(jù)按key做reduce，然后對各批次數(shù)據(jù)累加
WordCount的updateStateByKey版本

transform(func)
通過對原 DStream 的每個(gè) RDD 應(yīng)用轉(zhuǎn)換函數(shù)，創(chuàng)建一個(gè)新的 DStream。
官方文檔代碼舉例

Window operations
窗口操作：基于 window 對數(shù)據(jù) transformation(個(gè)人認(rèn)為與Storm的tick相似，但功能更強(qiáng)大)。
參數(shù)：窗口長度(window length)和滑動時(shí)間間隔(slide interval)必須是源DStream 批次間隔的倍數(shù)。
舉例說明：窗口長度為3，滑動時(shí)間間隔為2;上一行是原始 DStream，下一行是窗口化的 DStream。

常見 window operation

官方文檔代碼舉例

join(otherStream, [numTasks])
連接數(shù)據(jù)流
官方文檔代碼舉例1

官方文檔代碼舉例2

Output Operation

緩存與持久化：
通過 persist()將 DStream 中每個(gè) RDD 存儲在內(nèi)存。
Window operations 會自動持久化在內(nèi)存，無需顯示調(diào)用 persist()。
通過網(wǎng)絡(luò)接收的數(shù)據(jù)流(如Kafka、Flume、Socket、ZeroMQ、RocketMQ等)執(zhí)行 persist()時(shí)，默認(rèn)在兩個(gè)節(jié)點(diǎn)上持久化序列化后的數(shù)據(jù)，實(shí)現(xiàn)容錯。
Checkpoint：
用途：Spark 基于容錯存儲系統(tǒng)(如HDFS、S3)進(jìn)行故障恢復(fù)。
分類：
元數(shù)據(jù)檢查點(diǎn)：保存流式計(jì)算信息用于 Driver 運(yùn)行節(jié)點(diǎn)的故障恢復(fù)，包括創(chuàng)建應(yīng)用程序的配置、應(yīng)用程序定義的 DStream operations、已入隊(duì)但未完成的批次。
數(shù)據(jù)檢查點(diǎn)：保存生成的 RDD。由于 stateful transformation 需要合并多個(gè)批次的數(shù)據(jù)，即生成的 RDD 依賴于前幾個(gè)批次 RDD 的數(shù)據(jù)(dependency chain)，為縮短 dependency chain 從而減少故障恢復(fù)時(shí)間，需將中間 RDD 定期保存至可靠存儲(如HDFS)。
使用時(shí)機(jī)：
Stateful transformation：updateStateByKey()以及 window operations。
需要 Driver 故障恢復(fù)的應(yīng)用程序。
003、使用方法
Stateful transformation

需要 Driver 故障恢復(fù)的應(yīng)用程序(以WordCount舉例)：如果 checkpoint 目錄存在，則根據(jù) checkpoint 數(shù)據(jù)創(chuàng)建新 StreamingContext;否則(如首次運(yùn)行)新建 StreamingContext。

checkpoint 時(shí)間間隔
方法：

原則：一般設(shè)置為滑動時(shí)間間隔的5-10倍。
分析：checkpoint 會增加存儲開銷、增加批次處理時(shí)間。當(dāng)批次間隔較小(如1秒)時(shí)，checkpoint 可能會減小 operation 吞吐量;反之，checkpoint 時(shí)間間隔較大會導(dǎo)致 lineage 和 task 數(shù)量增長。
004、性能調(diào)優(yōu)
降低批次處理時(shí)間：
數(shù)據(jù)接收并行度
增加 DStream：接收網(wǎng)絡(luò)數(shù)據(jù)(如Kafka、Flume、Socket等)時(shí)會對數(shù)據(jù)反序列化再存儲在 Spark，由于一個(gè) DStream 只有 Receiver 對象，如果成為瓶頸可考慮增加 DStream。

設(shè)置“spark.streaming.blockInterval”參數(shù)：接收的數(shù)據(jù)被存儲在 Spark 內(nèi)存前，會被合并成 block，而 block 數(shù)量決定了Task數(shù)量;舉例，當(dāng)批次時(shí)間間隔為2秒且 block 時(shí)間間隔為200毫秒時(shí)，Task 數(shù)量約為10;如果Task數(shù)量過低，則浪費(fèi)了 CPU 資源;推薦的最小block時(shí)間間隔為50毫秒。
顯式對 Input DStream 重新分區(qū)：在進(jìn)行更深層次處理前，先對輸入數(shù)據(jù)重新分區(qū)。

數(shù)據(jù)處理并行度：reduceByKey、reduceByKeyAndWindow 等 operation 可通過設(shè)置“spark.default.parallelism”參數(shù)或顯式設(shè)置并行度方法參數(shù)控制。
數(shù)據(jù)序列化：可配置更高效的 Kryo 序列化。
設(shè)置合理批次時(shí)間間隔
原則：處理數(shù)據(jù)的速度應(yīng)大于或等于數(shù)據(jù)輸入的速度，即批次處理時(shí)間大于或等于批次時(shí)間間隔。
方法：
先設(shè)置批次時(shí)間間隔為5-10秒以降低數(shù)據(jù)輸入速度;
再通過查看 log4j 日志中的“Total delay”，逐步調(diào)整批次時(shí)間間隔，保證“Total delay”小于批次時(shí)間間隔。
內(nèi)存調(diào)優(yōu)
持久化級別：開啟壓縮，設(shè)置參數(shù)“spark.rdd.compress”。
GC策略：在Driver和Executor上開啟CMS。