micro-batches Processing & Continuous Processing

Structured Streaming 在Apache Spark 2.0引入，計(jì)算模式就是小批量計(jì)算，從高層次上看起來和小批量處理沒有什么關(guān)系的，主要有兩個(gè)原因。第一：開發(fā)者編程更簡(jiǎn)單，接口調(diào)用不需要關(guān)注小批量。第二：允許開發(fā)者可以把源源不斷的數(shù)據(jù)流看做一張無界的表，在發(fā)起查詢的時(shí)候就是靜態(tài)的表了。

spark 2.3中引入一種能夠達(dá)到毫秒級(jí)低延遲的計(jì)算模式：持續(xù)計(jì)算。
兩種計(jì)算模式如下：默認(rèn)（micro-batches）

圖片.png

micro-batches Processing：

使用：

 .filter("isPaymentFlagged(paymentId)") 

 .writeStream 

 {...}

 .trigger(processingTime = "0 seconds") 

 .start()

延遲性：

最低100 ms

圖片.png

原理：

在小批量處理模式下，spark streaming 計(jì)算引擎階段性地檢查數(shù)據(jù)流，然后批量處理數(shù)據(jù)，high-level 上的流程圖

圖片.png

在處理一批數(shù)據(jù)之前，先把這一批數(shù)據(jù)記錄的偏移量寫到whl日志中（write head log）（用于下一批數(shù)據(jù)查詢）, 等到把偏移量保存完成后開始計(jì)算，這樣就產(chǎn)生了延遲，從數(shù)據(jù)記錄的level上流程圖如下：

圖片.png

Continuous Processing：

使用：

.filter("isPaymentFlagged(paymentId)") 

 .writeStream \

 {...}

 .trigger(continuous = "5 seconds") 

 .start

延遲分析：

最低1 ms以下

圖片.png

原理：

在持續(xù)計(jì)算模式下：不是階段性的發(fā)起task，而是spark發(fā)起一個(gè)長(zhǎng)期運(yùn)行的long-running task，持續(xù)地讀、計(jì)算、寫。high-level流程圖如下：而對(duì)于保存數(shù)據(jù)記錄的偏移量，則是相當(dāng)于在數(shù)據(jù)流流入spark的時(shí)候上打標(biāo)記，兩個(gè)標(biāo)記之間叫 epoch，跟階段的意思差不多，task在遇到一個(gè)標(biāo)記的時(shí)候會(huì)異步的保存這個(gè)偏移量，對(duì)于持續(xù)計(jì)算是沒有影響的。

圖片.png

圖片.png

后記：

• 如果你對(duì)延遲性要求比較高的話可以用Continuous Processing 模式，而 micro-batches Processing 模式的吞吐量會(huì)更高。
• 持續(xù)計(jì)算在2.3中引入的，還是實(shí)驗(yàn)性的

@轉(zhuǎn)載原創(chuàng)文章 請(qǐng)標(biāo)明出處