Spark Streaming數據限流簡述

来源:https://www.cnblogs.com/softlin/archive/2020/01/19/12215446.html

  Spark Streaming對實時數據流進行分析處理,源源不斷的從數據源接收數據切割成一個個時間間隔進行處理;    流處理與批處理有明顯區別,批處理中的數據有明顯的邊界、數據規模已知;而流處理數據流並沒有邊界,也未知數據規模;    ...


  Spark Streaming對實時數據流進行分析處理,源源不斷的從數據源接收數據切割成一個個時間間隔進行處理;
  流處理與批處理有明顯區別,批處理中的數據有明顯的邊界、數據規模已知;而流處理數據流並沒有邊界,也未知數據規模;
  由於流處理的數據流特征,使之數據流具有不可預測性,而且數據處理的速率還與硬體、網路等資源有關,在這種情況下如不對源源不斷進來的數據流速率進行限制,那當Spark節點故障、網路故障或數據處理吞吐量下來時還有數據不斷流進來,那將有可能將出現OOM進而導致Spark Streaming程式崩潰;
  在Spark Streaming中不同的數據源採用不同的限速策略,但無論是Socket數據源的限流策略還是Kafka數據源的限流策略其速率(rate)的計算都是使用PIDController演算法進行計算而得來;
  下麵從源碼的角度分別介紹Socket數據源Kafka數據源的限流處理。

速率限制的計算與更新

  Spark Streaming的流處理其實是基於微批處理(MicroBatch)的,也就是說將數據流按某比較小的時間間隔將數據切割成為一段段微批數據進行處理;

  StreamingContext調用Start()啟動的時候會將速率控制器(rateController)添加到StreamingListener監聽器中;
  當每批次處理完成時將觸發監聽器(RateController),使用該批處理的處理結束時間、處理延遲時間、調度延遲時間、記錄行數調用PIDRateEstimator傳入PID演算法中(PID Controller)計算出該批次的速率(rate)並更新速率限制(rateLimit)與發佈該限制速率;

override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted) {
val elements = batchCompleted.batchInfo.streamIdToInputInfo

for {
  processingEnd <- batchCompleted.batchInfo.processingEndTime
  workDelay <- batchCompleted.batchInfo.processingDelay
  waitDelay <- batchCompleted.batchInfo.schedulingDelay
  elems <- elements.get(streamUID).map(_.numRecords)
 } computeAndPublish(processingEnd, elems, workDelay, waitDelay)
}

private def computeAndPublish(time: Long, elems: Long, workDelay: Long, waitDelay: Long): Unit =
Future[Unit] {
  val newRate = rateEstimator.compute(time, elems, workDelay, waitDelay)
  newRate.foreach { s =>
    rateLimit.set(s.toLong)
    publish(getLatestRate())
  }
}

Socket數據源限流

  批次的限制速率上面已經算出,這裡說的是接收Socket過來的數據時的數據限流;
  SocketInputStream類receive方法接收到數據後將數據存入 BlockGenerator的Buffer中,在寫入Buffer前調用限流器 (RateLimiter)對寫入數據進行限流;
  RateLimiter限流器使用了Google開源的 Guava中內置的RateLimiter限流器,該類只是對Guava限流器的簡單封裝;
  在Spark Streaming中可通過使用兩個參數配置初始速率與最大速率spark.streaming.receiver.maxRate、spark.streaming.backpressure.initialRate;亦可配置PIDController演算法相關的四個參數值;
  RateLimiter限流器是基於令牌桶的演算法基本原理比較簡單,以一個恆定的速率生成令牌放入令牌桶中,桶滿則停止,處理請求時需要從令牌桶中取出令牌,當桶中無令牌可取時阻塞等待,此演算法用於確保系統不被洪峰擊垮。

  private lazy val rateLimiter = GuavaRateLimiter.create(getInitialRateLimit().toDouble)
/**
  * Push a single data item into the buffer.
 */
 def addData(data: Any): Unit = {
  if (state == Active) {
//調用限流器等待
  waitToPush()
  synchronized {
    if (state == Active) {
      currentBuffer += data
    } else {
      throw new SparkException(
        "Cannot add data as BlockGenerator has not been started or has been stopped")
    }
  }
} else {
  throw new SparkException(
    "Cannot add data as BlockGenerator has not been started or has been stopped")
 }
}

def waitToPush() {
   //限流器申請令牌
   rateLimiter.acquire()
}

Guava庫中RateLimiter限流器基本使用:

//創建限流器,每秒產生令牌數1
    RateLimiter rateLimiter=RateLimiter.create(1);
    for (int i = 0; i < 10; i++) {
        //獲得一個令牌,未申請到令牌則阻塞等待
        double waitTime = rateLimiter.acquire();
        System.out.println(String.format("id:%d time:%d waitTime:%f",i,System.currentTimeMillis(),waitTime));
    }

Kafka數據源限流的實現

  在Spark Streaming Kafka包拉取Kafka數據會進行如下動作:
  1、取Kafka中最新偏移量、分區
  2、通過rateController限制每個分區可拉取的最大消息數
  3、在DirectKafkaInputDStream中創建KafkaRDD,在其中調用相關對象拉取數據

  通過如上步驟也可用看出,只要限制了Kafka某個分區的偏移量(offset)範圍也就可限制從Kafka拉取的消息數量,從而達到限流的目的,Spark streaming kafka也是通過此實現的;

計算每個分區速率限制,有如下步驟:
  1、通過seekToEnd獲取最新可用偏移量與當前偏移量對比獲得當前所有分區延遲偏移量
  單個分區偏移量延遲=最新偏移量記錄-當前偏移量記錄
  2、獲取配置項中每個分區最大速率
(spark.streaming.kafka.maxRatePerPartition),背壓率計算,計算每個分區背壓率計算公式為:
  單個分區背壓率=單個分區偏移量延遲/所有分區總延遲*速率限制
  速率限制(rateLimit):為通過PIDController動態計算得來

  如有配置每個分區最大速率則取配置項最大速率與背壓率兩者中的最小值未配置則取背壓率作為每個分區速率限制;

  3、將批次間隔(batchDuration)*每個分區速率限制=每個分區最大消息數
  4、取當前分區偏移量+分區最大消息數 與 最新偏移量兩者當中最小的,由此來控制拉取消息速率;

  如當前偏移量+分區最大消息數 大於 最新偏移量則取 最新偏移量否則取 當前偏移量+分區最大消息數作為拉取Kafka數據的Offset範圍

// 限制每個分區最大消息數
protected def clamp(
offsets: Map[TopicPartition, Long]): Map[TopicPartition, Long] = {

maxMessagesPerPartition(offsets).map { mmp =>
  mmp.map { case (tp, messages) =>
      val uo = offsets(tp)
      tp -> Math.min(currentOffsets(tp) + messages, uo)
  }
}.getOrElse(offsets)
}

  不管是Kafka數據源還是Socket數據源Spark Streaming中都使用了PIDController演算法用於計算其速率限制值,兩者的差別也只是因為兩種數據源的獲取方式數據特征而決定的。Socket數據源使用了Guava RateLimiter、而Kafka數據源自己實現了基於Offsets的限流
  以上說介紹的框架版本為:Spark Streaming 版本為2.3.2與spark-streaming-kafka-0-10_2.11;

參考資料:
http://kafka.apache.org
http://spark.apache.org

文章首發地址:Solinx
https://mp.weixin.qq.com/s/yHStZgTAGBPoOMpj4e27Jg


您的分享是我們最大的動力!

更多相關文章
  • 本文是針對我的工具 "藍奏雲批量下載工具" 的補充說明筆記,準備按照流程整理我實現軟體的思路與方法。 涉及知識 Java的IO流 Java的下載文件 HtmlUnit的使用方法 okhttp的使用 分析與軟體思路 在某一天,我找到了一部電子書的資源,但是,該藍奏雲地址是一個文件夾,由於藍奏雲不支持批 ...
  • Java 泛型通配符 ? extends super 的用法 示例 1 : ? extends ArrayList heroList 表示這是一個Hero泛型或者其子類泛型 heroList 的泛型可能是Hero heroList 的泛型可能是APHero heroList 的泛型可能是ADHero ...
  • 1. 欄位查詢 通過模型類.objects屬性可以調用如下函數,實現對模型類對應的數據表的查詢。 函數名 功能 返回值 說明 get 返回表中滿足條件的一條且只能有一條數據。 返回值是一個模型類對象。 參數中寫查詢條件。 1)如果查到多條數據,則拋異常MultipleObjectsReturned。 ...
  • 寫在前面 首先,祝大家新年快樂,在國人的情結里,現在才是真正的年底,估計現在好多朋友已經陸續回家過節去了,祝回家的路順風。過年回去就好好陪伴家人,不要看一些毒瘤號寫的,過年時間彎道超越別人,趁著假期學習什麼各種新知識,這完全瞎扯,這一年的時間該努力,努力了,吃苦也吃了,春節就好好陪伴家人吃喝玩樂。 ...
  • Python的駐留機制及為在同一運行空間內,當兩變數的值相同,則地址也相同。 舉例: 以上示例為駐留機制有效的情況下的記憶體地址。 以下為加入非數字、字母、下劃線內容的駐留機制驗證,結果為無效 ...
  • 在經常性讀取大量的數值文件時(比如深度學習訓練數據),可以考慮現將數據存儲為Numpy格式,然後直接使用Numpy去讀取,速度相比為轉化前快很多. 下麵就常用的保存數據到二進位文件和保存數據到文本文件進行介紹: 1.保存為二進位文件(.npy/.npz) numpy.save 保存一個數組到一個二進 ...
  • list list(列表)是Python內置的一種數據類型,它是一種有序、可變的集合,可以隨時添加和刪除其中的元素。 變數classmates就是一個list。關於list的操作如下: 方法 len():獲取list元素的個數 通過索引訪問 append:向list末尾追加元素 insert:把元素 ...
  • @ "TOC" JDK1.7:數組+鏈表 JDK1.8:數組+鏈表+紅黑樹 前五個問題環境用的是是JDK1.7,後面全部是1.8 1、Hash的計算規則? 簡單的說是個“擾動函數”,目的是為了使散列分佈的更加均勻。 具體演算法是用key的Hashcode值右移16位,將hashcode高位和低位的值進 ...
一周排行
  • 前幾天發佈了 "抄抄《CSS 故障藝術》的動畫" 這篇文章,在這篇文章里介紹瞭如何使用Win2D繪製文字然後配合BlendEffect製作故障藝術的動畫。本來打算就這樣收手不玩這個動畫了,但後來又發現性能不符合理想。明明只是做做Resize動畫和用BlendEffect混合,為什麼性能會這麼差呢? ...
  • 控制條控制項: progressBar 不能按照你程式的進程自動變化,需認為計算,調整變化量 private void progressBar1_Click(object sender, EventArgs e) { this.progressBar1.Maximum = 100;//設置進度條最大長 ...
  • 首先創建一個asp.net core web應用程式 第二步 目前官方預置了7種模板項目供我們選擇。從中我們可以看出,既有我們熟悉的MVC、WebAPI,又新添加了Razor Page,以及結合比較流行的Angular、React前端框架的模板項目。 空項目模板 Program.cs using S ...
  • 對閉包的理解 1.對於成員變數和局部變數:成員變數就是方法外部,類的內部定義的變數;局部變數就是方法或語句塊內部定義的變數。局部變數必須初始化。 形式參數是局部變數,局部變數的數據存在於棧記憶體中。棧記憶體中的局部變數隨著方法的消失而消失。成員變數存儲在堆中的對象裡面,由垃圾回收器負責回收。 成員變數它 ...
  • Xamarin.Forms讀取並展示Android和iOS通訊錄 TerminalMACS客戶端 本文同步更新地址: https://dotnet9.com/11520.html https://terminalmacs.com/861.html 閱讀導航: 一、功能說明 二、代碼實現 三、源碼獲取 ...
  • 做下對文件複製操作相關的筆記: /// <summary> /// 文件幫助類 /// </summary> public class FileHelper { /// <summary> /// 複製一個目錄下所有文件到一個新目錄下 /// </summary> /// <param name=" ...
  • 前言 有一個東西叫做鴨子類型,所謂鴨子類型就是,只要一個東西表現得像鴨子那麼就能推出這玩意就是鴨子。 C 裡面其實也暗藏了很多類似鴨子類型的東西,但是很多開發者並不知道,因此也就沒法好好利用這些東西,那麼今天我細數一下這些藏在編譯器中的細節。 不是只有 和 才能 在 C 中編寫非同步代碼的時候,我們經 ...
  • [toc] 1.應用背景 底端設備有大量網路報文(位元組數組):心跳報文,數據採集報文,告警報文上報。需要有對應的報文結構去解析這些位元組流數據。 2.結構體解析 由此,我第一點就想到了用結構體去解析。原因有以下兩點: 2.1.結構體存在棧中 類屬於引用類型,存在堆中;結構體屬於值類型,存在棧中,在一個 ...
  • 《深入淺出 C#》 (第3版) [作者] (美) Andrew Stellman (美) Jennifer Greene[譯者] (中) 徐陽 丁小峰 等譯[出版] 中國電力出版社[版次] 2016年08月 第1版[印次] 2018年04月 第4次 印刷[定價] 148.00元 【引子】 要學習編程 ...
  • 記錄使用對象初始值設定項初始化對象。 using System; using System.Collections.Generic; namespace ConsoleApp2 { class Program { static void Main(string[] args) { // 使用構造函數 ...
x