需要關(guān)于Spark Streaming相關(guān)學(xué)習(xí)資料視頻可以加QQ：1653978901

?

Spark Streaming 支持實(shí)時(shí)數(shù)據(jù)流的可擴(kuò)展(Scalable)、高吞吐(high-throughput)、容錯(cuò)(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)容錯(cuò)。

Checkpoint：

用途：Spark 基于容錯(cuò)存儲系統(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。

總結(jié)

以上是生活随笔為你收集整理的Spark Streaming 技术看点！的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。