Spark Streaming 反壓機(jī)制是1.5版本推出的特性，用來(lái)解決處理速度比攝入速度慢的情況，簡(jiǎn)單來(lái)講就是做流量控制。當(dāng)批處理時(shí)間(Batch Processing Time)大于批次間隔(Batch Interval，即 BatchDuration)時(shí),說(shuō)明處理數(shù)據(jù)的速度小于數(shù)據(jù)攝入的速度，持續(xù)時(shí)間過(guò)長(zhǎng)或源頭數(shù)據(jù)暴增，容易造成數(shù)據(jù)在內(nèi)存中堆積，最終導(dǎo)致Executor OOM。反壓就是來(lái)解決這個(gè)問(wèn)題的。

　　spark streaming的消費(fèi)數(shù)據(jù)源方式有兩種：

　　若是基于Receiver的數(shù)據(jù)源，可以通過(guò)設(shè)置spark.streaming.receiver.maxRate來(lái)控制最大輸入速率;

　　若是基于Direct的數(shù)據(jù)源(如Kafka Direct Stream)，則可以通過(guò)設(shè)置spark.streaming.kafka.maxRatePerPartition來(lái)控制最大輸入速率。

　　當(dāng)然，在事先經(jīng)過(guò)壓測(cè)，且流量高峰不會(huì)超過(guò)預(yù)期的情況下，設(shè)置這些參數(shù)一般沒(méi)什么問(wèn)題。但最大值，不代表是最優(yōu)值，最好還能根據(jù)每個(gè)批次處理情況來(lái)動(dòng)態(tài)預(yù)估下個(gè)批次最優(yōu)速率。

　　在Spark 1.5.0以上，就可通過(guò)背壓機(jī)制來(lái)實(shí)現(xiàn)。開(kāi)啟反壓機(jī)制，即設(shè)置spark.streaming.backpressure.enabled為true，Spark Streaming會(huì)自動(dòng)根據(jù)處理能力來(lái)調(diào)整輸入速率，從而在流量高峰時(shí)仍能保證最大的吞吐和性能

　　Spark Streaming的反壓機(jī)制中，有以下幾個(gè)重要的組件：

　　RateController

　　RateController 組件是 JobScheduler 的監(jiān)聽(tīng)器，主要監(jiān)聽(tīng)集群所有作業(yè)的提交、運(yùn)行、完成情況，并從 BatchInfo 實(shí)例中獲取以下信息，交給速率估算器(RateEstimator)做速率的估算。

　　當(dāng)前批次任務(wù)處理完成的時(shí)間戳 (processingEndTime)

　　該批次從第一個(gè) job 到最后一個(gè) job 的實(shí)際處理時(shí)長(zhǎng) (processingDelay)

　　該批次的調(diào)度時(shí)延，即從被提交到 JobScheduler 到第一個(gè) job 開(kāi)始處理的時(shí)長(zhǎng)(schedulingDelay)

　　該批次輸入數(shù)據(jù)的總條數(shù)(numRecords)

　　RateEstimator

　　Spark 2.x 只支持基于 PID 的速率估算器，這里只討論這種實(shí)現(xiàn)。基于 PID 的速率估算器簡(jiǎn)單地說(shuō)就是它把收集到的數(shù)據(jù)(當(dāng)前批次速率)和一個(gè)設(shè)定值(上一批次速率)進(jìn)行比較，然后用它們之間的差計(jì)算新的輸入值，估算出一個(gè)合適的用于下一批次的流量閾值。這里估算出來(lái)的值就是流量的閾值，用于更新每秒能夠處理的最大記錄數(shù)

　　RateLimiter

　　以上這兩個(gè)組件都是在Driver端用于更新最大速度的，而RateLimiter是用于接收到Driver的更新通知之后更新Executor的最大處理速率的組件。RateLimiter是一個(gè)抽象類，它并不是Spark本身實(shí)現(xiàn)的，而是借助了第三方Google的GuavaRateLimiter來(lái)產(chǎn)生的。它實(shí)質(zhì)上是一個(gè)限流器，也可以叫做令牌，如果Executor中task每秒計(jì)算的速度大于該值則阻塞，如果小于該值則通過(guò)，將流數(shù)據(jù)加入緩存中進(jìn)行計(jì)算。

　　* 反壓機(jī)制真正起作用時(shí)需要至少處理一個(gè)批：由于反壓機(jī)制需要根據(jù)當(dāng)前批的速率，預(yù)估新批的速率，所以反壓機(jī)制真正起作用前，應(yīng)至少保證處理一個(gè)批。

　　* 如何保證反壓機(jī)制真正起作用前應(yīng)用不會(huì)崩潰：要保證反壓機(jī)制真正起作用前應(yīng)用不會(huì)崩潰,需要控制每個(gè)批次最大攝入速率。若為Direct Stream，如Kafka Direct Stream,則可以通過(guò)spark.streaming.kafka.maxRatePerPartition參數(shù)來(lái)控制。此參數(shù)代表了 每秒每個(gè)分區(qū)最大攝入的數(shù)據(jù)條數(shù)。假設(shè)BatchDuration為10秒,spark.streaming.kafka.maxRatePerPartition為12條,kafka topic 分區(qū)數(shù)為3個(gè)，則一個(gè)批(Batch)最大讀取的數(shù)據(jù)條數(shù)為360條(3*12*10=360)。同時(shí)，需要注意，該參數(shù)也代表了整個(gè)應(yīng)用生命周期中的最大速率，即使是背壓調(diào)整的最大值也不會(huì)超過(guò)該參數(shù)。

　　反壓相關(guān)的參數(shù)

　　更多關(guān)于大數(shù)據(jù)培訓(xùn)的問(wèn)題，歡迎咨詢千鋒教育在線名師，如果想要了解我們的師資、課程、項(xiàng)目實(shí)操的話可以點(diǎn)擊咨詢課程顧問(wèn)，獲取試聽(tīng)資格來(lái)試聽(tīng)我們的課程，在線零距離接觸千鋒教育大咖名師，讓你輕松從入門到精通。

　　注：本文部分文字和圖片來(lái)源于網(wǎng)絡(luò)，如有侵權(quán)，請(qǐng)聯(lián)系刪除。版權(quán)歸原作者所有!