關(guān)注公眾號(hào)：大數(shù)據(jù)技術(shù)派，回復(fù)“資料”，領(lǐng)取1000G資料。
本文首發(fā)于我的個(gè)人博客：Flink 是如何統(tǒng)一批流引擎的

2015 年，F(xiàn)link 的作者就寫了 Apache Flink: Stream and Batch Processing in a Single Engine 這篇論文。本文以這篇論文為引導(dǎo)，詳細(xì)講講 Flink 內(nèi)部是如何設(shè)計(jì)并實(shí)現(xiàn)批流一體的架構(gòu)。

前言

通常我們?cè)?Flink 中說(shuō)批流一體指的是這四個(gè)方向，其中 Runtime 便是 Flink 運(yùn)行時(shí)的實(shí)現(xiàn)。

數(shù)據(jù)交換模型

Flink 對(duì)于流作業(yè)和批作業(yè)有一個(gè)統(tǒng)一的執(zhí)行模型。

Flink 中每個(gè) Task 的輸出會(huì)以 IntermediateResult 做封裝，內(nèi)部并沒(méi)有對(duì)流和批兩種作業(yè)做一個(gè)明確的劃分，只是通過(guò)不同類型的 IntermediateResult 來(lái)表達(dá) PIPELINED 和 BLOCKING 這兩大類數(shù)據(jù)交換模型。

在了解數(shù)據(jù)交換模型之前，我們來(lái)看下為什么 Flink 對(duì)作業(yè)類型不作區(qū)分，這樣的好處是什么？

如上圖所示，假如我們有一個(gè)工作需要將批式作業(yè)執(zhí)行結(jié)果作為流式作業(yè)的啟動(dòng)輸入，那怎么辦？這個(gè)作業(yè)是算批作業(yè)還是流作業(yè)？

很顯然，以我們的常識(shí)是無(wú)法定義的，而現(xiàn)有的工業(yè)界的辦法也是如此，將這個(gè)作業(yè)拆分為兩個(gè)作業(yè)，先跑批式作業(yè)，再跑流式作業(yè)，這樣當(dāng)然可以，但是人工運(yùn)維的成本也是足夠大的：

  1. 需要一個(gè)外界存儲(chǔ)來(lái)管理批作業(yè)的輸出數(shù)據(jù)。
  2. 需要一個(gè)支持批流作業(yè)依賴的調(diào)度系統(tǒng)。

如果期望實(shí)現(xiàn)這樣一個(gè)作業(yè)，那么首先執(zhí)行這個(gè)作業(yè)的計(jì)算引擎的作業(yè)屬性就不能對(duì)批作業(yè)和流作業(yè)進(jìn)行強(qiáng)綁定。那么 Flink 能否實(shí)現(xiàn)這樣的需求呢？我們先來(lái)看看數(shù)據(jù)交換的具體細(xì)節(jié)，最后再來(lái)一起看看這個(gè)作業(yè)的可行性。

我們以 PIPELINED 數(shù)據(jù)交換模型為例，看看是如何設(shè)計(jì)的：

PIPELINED 模式下，RecordWriter 將數(shù)據(jù)放入到 Buffer 中，根據(jù) Key 的路由規(guī)則發(fā)送給對(duì)應(yīng)的 Partition，Partition 將自己的數(shù)據(jù)封裝到 Reader 中放入隊(duì)列，讓 Netty Server 從隊(duì)列中讀取數(shù)據(jù)，發(fā)送給下游。

我們將數(shù)據(jù)交換模式改為 BLOCKING，會(huì)發(fā)現(xiàn)這個(gè)設(shè)計(jì)也是同樣可行的。Partition 將數(shù)據(jù)寫入到文件，而 Reader 中維護(hù)著文件的句柄，上游任務(wù)結(jié)束后調(diào)度下游任務(wù)，而下游任務(wù)通過(guò) Netty Client 的 Partition Request 喚醒對(duì)應(yīng)的 Partition 和 Reader，將數(shù)據(jù)拉到下游。

調(diào)度模型

有 LAZY 和 EAGER 兩種調(diào)度模型，默認(rèn)情況下流作業(yè)使用 EAGER，批作業(yè)使用 LAZY。

EAGER

這個(gè)很好理解，因?yàn)榱魇阶鳂I(yè)是 All or Nothing 的設(shè)計(jì)，要么所有 Task 都 Run 起來(lái)，要么就不跑。

LAZY

LAZY 模式就是先調(diào)度上游，等待上游產(chǎn)生數(shù)據(jù)或結(jié)束后再調(diào)度下游。有些類似 Spark 中的 Stage 執(zhí)行模式。

Region Scheduling

可以看到，不管是 EAGER 還是 LAZY 都沒(méi)有辦法執(zhí)行我們剛才提出的批流混合的任務(wù)，于是社區(qū)提出了 Region Scheduling 來(lái)統(tǒng)一批流作業(yè)的調(diào)度，我們先看一下如何定義 Region：

以 Join 算子為例，我們都知道如果 Join 算子的兩個(gè)輸入都是海量數(shù)據(jù)的話，那么我們是需要等兩個(gè)輸入的數(shù)據(jù)都完全準(zhǔn)備好才能進(jìn)行 Join 操作的，所以 Join 兩條輸入的邊對(duì)應(yīng)的數(shù)據(jù)交換模式對(duì)應(yīng)的應(yīng)該是 BLOCKING 模式，我們可以根據(jù) BLOCKING 的邊將作業(yè)劃分為多個(gè)子 Region，如上圖虛線所示。

如果實(shí)現(xiàn)了 Region Scheduling 之后，我們?cè)谏厦嫣岬降呐骰旌系淖鳂I(yè)就可以將深色部分流式作業(yè)劃為一個(gè) Region，淺色部分批式作業(yè)再劃分為多個(gè) Region，而淺色部分是深色部分 Region 的輸入，所以根據(jù) Region Scheduling 的原則會(huì)優(yōu)先調(diào)度最前面的 Region。

總結(jié)

上面提到了數(shù)據(jù)交換模型和調(diào)度模型，簡(jiǎn)單來(lái)講其實(shí)就兩句話：

1 實(shí)現(xiàn)了用 PIPELINED 模型去跑批式作業(yè)

用 PIPELINED 模型跑流式作業(yè)和用 BLOCKING 模型跑批式作業(yè)都是沒(méi)有什么新奇的。這里提到用 PIPELINED 模式跑批作業(yè)，主要是針對(duì)實(shí)時(shí)分析的場(chǎng)景，以 Spark 為例，在大部分出現(xiàn) Shuffle 或是聚合的場(chǎng)景下都會(huì)出現(xiàn)落盤的行為，并且調(diào)度順序是一個(gè)一個(gè) Stage 進(jìn)行調(diào)度，極大地降低了數(shù)據(jù)處理的實(shí)時(shí)性，而使用 PIPELINED 模式會(huì)對(duì)性能有一定提升。

可能有人會(huì)問(wèn)類似 Join 的算子如何使用 PIPELINED 數(shù)據(jù)交換模型實(shí)現(xiàn)不落盤的操作？事實(shí)上 Flink 也會(huì)落盤，只不過(guò)不是在 Join 的兩個(gè)輸入端落盤，而是將兩個(gè)輸入端的數(shù)據(jù)傳輸?shù)?Join 算子上，內(nèi)存撐不住時(shí)再進(jìn)行落盤，海量數(shù)據(jù)下和 Spark 的行為并沒(méi)有本質(zhì)區(qū)別，但是在數(shù)據(jù)量中等，內(nèi)存可容納的情況下會(huì)帶來(lái)很大的收益。

2 集成了一部分調(diào)度系統(tǒng)的功能

根據(jù) Region 來(lái)調(diào)度作業(yè)時(shí)，Region 內(nèi)部跑的具體是流作業(yè)還是批作業(yè)，F(xiàn)link 自身是不關(guān)心的，更關(guān)心的 Region 之間的依賴關(guān)系，一定程度上，利用這種調(diào)度模型我們可以將過(guò)去需要拆分為多個(gè)作業(yè)的執(zhí)行模式放到一個(gè)作業(yè)中來(lái)執(zhí)行，比如上面提到的批流混合的作業(yè)。