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

在成都網(wǎng)站建設(shè)、網(wǎng)站設(shè)計(jì)中從網(wǎng)站色彩、結(jié)構(gòu)布局、欄目設(shè)置、關(guān)鍵詞群組等細(xì)微處著手，突出企業(yè)的產(chǎn)品/服務(wù)/品牌，幫助企業(yè)鎖定精準(zhǔn)用戶，提高在線咨詢和轉(zhuǎn)化，使成都網(wǎng)站營(yíng)銷成為有效果、有回報(bào)的無(wú)錫營(yíng)銷推廣。創(chuàng)新互聯(lián)專業(yè)成都網(wǎng)站建設(shè)10余年了，客戶滿意度97.8%，歡迎成都創(chuàng)新互聯(lián)客戶聯(lián)系。

架構(gòu)圖
特性如下：
? 可線性伸縮至超過(guò)數(shù)百個(gè)節(jié)點(diǎn);
? 實(shí)現(xiàn)亞秒級(jí)延遲處理;
? 可與 Spark 批處理和交互式處理無(wú)縫集成;
? 提供簡(jiǎn)單的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 提供的高級(jí)抽象連續(xù)數(shù)據(jù)流。
組成：一個(gè) DStream 可看作一個(gè) RDDs 序列。
核心思想：將計(jì)算作為一系列較小時(shí)間間隔的、狀態(tài)無(wú)關(guān)的、確定批次的任務(wù)，每個(gè)時(shí)間間隔內(nèi)接收的輸入數(shù)據(jù)被可靠存儲(chǔ)在集群中，作為一個(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ù)源)和高級(jí)輸入源(Kafka、Flume等)。
Receiver：
每個(gè) Input DStream(文件流除外)都會(huì)對(duì)應(yīng)一個(gè)單一的 Receiver對(duì)象，負(fù)責(zé)從數(shù)據(jù)源接收數(shù)據(jù)并存入 Spark 內(nèi)存進(jìn)行處理。應(yīng)用程序中可創(chuàng)建多個(gè) Input DStream 并行接收多個(gè)數(shù)據(jù)流。
每個(gè) Receiver 是一個(gè)長(zhǎng)期運(yùn)行在Worker或者 Executor 上的 Task，所以會(huì)占用該應(yīng)用程序的一個(gè)核(core)。如果分配給 Spark Streaming 應(yīng)用程序的核數(shù)小于或等于 Input DStream 個(gè)數(shù)(即Receiver個(gè)數(shù))，則只能接收數(shù)據(jù)，卻沒(méi)有能力全部處理(文件流除外，因?yàn)闊o(wú)需Receiver)。
Spark Streaming 已封裝各種數(shù)據(jù)源，需要時(shí)參考官方文檔。
Transformation Operation
常用Transformation

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

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

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

常見 window operation

官方文檔代碼舉例

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

官方文檔代碼舉例2

Output Operation

緩存與持久化：
通過(guò) persist()將 DStream 中每個(gè) RDD 存儲(chǔ)在內(nèi)存。
Window operations 會(huì)自動(dòng)持久化在內(nèi)存，無(wú)需顯示調(diào)用 persist()。
通過(guò)網(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ò)存儲(chǔ)系統(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 定期保存至可靠存儲(chǔ)(如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è)置為滑動(dòng)時(shí)間間隔的5-10倍。
分析：checkpoint 會(huì)增加存儲(chǔ)開銷、增加批次處理時(shí)間。當(dāng)批次間隔較小(如1秒)時(shí)，checkpoint 可能會(huì)減小 operation 吞吐量;反之，checkpoint 時(shí)間間隔較大會(huì)導(dǎo)致 lineage 和 task 數(shù)量增長(zhǎng)。
004、性能調(diào)優(yōu)
降低批次處理時(shí)間：
數(shù)據(jù)接收并行度
增加 DStream：接收網(wǎng)絡(luò)數(shù)據(jù)(如Kafka、Flume、Socket等)時(shí)會(huì)對(duì)數(shù)據(jù)反序列化再存儲(chǔ)在 Spark，由于一個(gè) DStream 只有 Receiver 對(duì)象，如果成為瓶頸可考慮增加 DStream。

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

數(shù)據(jù)處理并行度：reduceByKey、reduceByKeyAndWindow 等 operation 可通過(guò)設(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ù)輸入速度;
再通過(guò)查看 log4j 日志中的“Total delay”，逐步調(diào)整批次時(shí)間間隔，保證“Total delay”小于批次時(shí)間間隔。
內(nèi)存調(diào)優(yōu)
持久化級(jí)別：開啟壓縮，設(shè)置參數(shù)“spark.rdd.compress”。
GC策略：在Driver和Executor上開啟CMS。

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無(wú)理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國(guó)服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡(jiǎn)單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢(shì)，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場(chǎng)景需求。

當(dāng)前題目：SparkStreaming技術(shù)點(diǎn)匯總-創(chuàng)新互聯(lián)
當(dāng)前路徑：http://jinyejixie.com/article36/djeosg.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供自適應(yīng)網(wǎng)站、品牌網(wǎng)站制作、外貿(mào)網(wǎng)站建設(shè)、App設(shè)計(jì)、營(yíng)銷型網(wǎng)站建設(shè)、企業(yè)建站

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)