TableSQL API和Pyhton上相關(guān)的性能優(yōu)化

一、簡介

1.14 新版本原本規(guī)劃有 35 個比較重要的新特性以及優(yōu)化工作，目前已經(jīng)有 26 個工作完成；5 個任務(wù)不確定是否能準(zhǔn)時完成；另外 4 個特性由于時間或者本身設(shè)計上的原因，會放到后續(xù)版本完成。［1］

1.14 相對于歷屆版本來說，囊括的優(yōu)化和新增功能點其實并不算多。通過觀察發(fā)版的節(jié)奏可以發(fā)現(xiàn)，通常在 1-2 個大版本后都會發(fā)布一個變化稍微少一點的版本，主要目的是把一些特性穩(wěn)定下來。

1.14 版本就是這樣一個定位，我們稱之為質(zhì)量改進和維護的版本。這個版本預(yù)計 8 月 16 日停止新特性開發(fā)，可能在 9 月份能夠和大家正式見面，有興趣可以關(guān)注以下鏈接去跟蹤功能發(fā)布進度。

Wiki：https://cwiki.apache.org/confluence/display/FLINK/1.14+Release

Jira：https://issues.apache.org/jira/projects/FLINK/versions/12349614

［1］ 截至到 8 月 31 日，確定進入新版本的是 33 個，已全部完成。

二、流批一體

流批一體其實從 Flink 1.9 版本開始就受到持續(xù)的關(guān)注，它作為社區(qū) RoadMap 的重要組成部分，是大數(shù)據(jù)實時化必然的趨勢。但是另一方面，傳統(tǒng)離線的計算需求其實并不會被實時任務(wù)完全取代，而是會長期存在。

在實時和離線的需求同時存在的狀態(tài)下，以往的流批獨立技術(shù)方案存在著一些痛點，比如：

需要維護兩套系統(tǒng)，相應(yīng)的就需要兩組開發(fā)人員，人力的投入成本很高；

另外，兩套數(shù)據(jù)鏈路處理相似內(nèi)容帶來維護的風(fēng)險性和冗余；

最重要的一點是，如果流批使用的不是同一套數(shù)據(jù)處理系統(tǒng)，引擎本身差異可能會存在數(shù)據(jù)口徑不一致的問題，從而導(dǎo)致業(yè)務(wù)數(shù)據(jù)存在一定的誤差。這種誤差對于大數(shù)據(jù)分析會有比較大的影響。

在這樣的背景下，F(xiàn)link 社區(qū)認(rèn)定了實時離線一體化的技術(shù)路線是比較重要的技術(shù)趨勢和方向。

Flink 在過去的幾個版本中，在流批一體方面做了很多的工作。可以認(rèn)為 Flink 在引擎層面，API 層面和算子的執(zhí)行層面上做到了真正的流與批用同一套機制運行。但是在任務(wù)具體的執(zhí)行模式上會有 2 種不同的模式：

下圖是不同的執(zhí)行模式：

對于無限的數(shù)據(jù)流，統(tǒng)一采用了流的執(zhí)行模式。流的執(zhí)行模式指的是所有計算節(jié)點是通過 Pipeline 模式去連接的，Pipeline 是指上游和下游計算任務(wù)是同時運行的，隨著上游不斷產(chǎn)出數(shù)據(jù)，下游同時在不斷消費數(shù)據(jù)。這種全 Pipeline 的執(zhí)行方式可以：

通過 eventTime 表示數(shù)據(jù)是什么時候產(chǎn)生的；

通過 watermark 得知在哪個時間點，數(shù)據(jù)已經(jīng)到達(dá)了；

通過 state 來維護計算中間狀態(tài)；

通過 Checkpoint 做容錯的處理。

這兩種各有優(yōu)劣，可以根據(jù)作業(yè)的具體場景來進行選擇。

對于有限的數(shù)據(jù)集有 2 種執(zhí)行模式，我們可以把它看成一個有限的數(shù)據(jù)流去做處理，也可以把它看成批的執(zhí)行模式。批的執(zhí)行模式雖然也有 eventTime，但是對于 watermark 來說只支持正無窮。對數(shù)據(jù)和 state 排序后，它在任務(wù)的調(diào)度和 shuffle 上會有更多的選擇。

流批的執(zhí)行模式是有區(qū)別的，最主要的就是批的執(zhí)行模式會有落盤的中間過程，只有當(dāng)前面任務(wù)執(zhí)行完成，下游的任務(wù)才會觸發(fā)，這個容錯機制是通過 shuffle 進行容錯的。

這 2 者也各有各的執(zhí)行優(yōu)勢：

對于流的執(zhí)行模式來說，它沒有落盤的壓力，同時容錯是基于數(shù)據(jù)的分段，通過不斷對數(shù)據(jù)進行打點 Checkpoint 去保證斷點恢復(fù)；

然而在批處理上，因為要經(jīng)過 shuffle 落盤，所以對磁盤會有壓力。但是因為數(shù)據(jù)是經(jīng)過排序的，所以對于批來說，后續(xù)的計算效率可能會有一定的提升。同時，在執(zhí)行時候是經(jīng)過分段去執(zhí)行任務(wù)的，無需同時執(zhí)行；在容錯計算方面是根據(jù) stage 進行容錯。

Flink 1.14 的優(yōu)化點主要是針對在流的執(zhí)行模式下，如何去處理有限數(shù)據(jù)集。之前處理無限數(shù)據(jù)集，和現(xiàn)在處理有限數(shù)據(jù)集最大的區(qū)別在于引入了 “任務(wù)可能會結(jié)束” 的概念。這種情況下帶來了一些新的問題，如下圖：

■ 在流的執(zhí)行模式下的 Checkpoint 機制

對于無限流，它的 Checkpoint 是由所有的 source 節(jié)點進行觸發(fā)的，由 source 節(jié)點發(fā)送 Checkpoint Barrier ，當(dāng) Checkpoint Barrier 流過整個作業(yè)時候，同時會存儲當(dāng)前作業(yè)所有的 state 狀態(tài)。

而在有限流的 Checkpoint 機制中，Task 是有可能提早結(jié)束的。上游的 Task 有可能先處理完任務(wù)提早退出了，但下游的 Task 卻還在執(zhí)行中。在同一個 stage 不同并發(fā)下，有可能因為數(shù)據(jù)量不一致導(dǎo)致部分任務(wù)提早完成了。這種情況下，在后續(xù)的執(zhí)行作業(yè)中，如何進行 Checkpoint？

在 1.14 中，JobManager 動態(tài)根據(jù)當(dāng)前任務(wù)的執(zhí)行情況，去明確 Checkpoint Barrier 是從哪里開始觸發(fā)。同時在部分任務(wù)結(jié)束后，后續(xù)的 Checkpoint 只會保存仍在運行 Task 所對應(yīng)的 stage，通過這種方式能夠讓任務(wù)執(zhí)行完成后，還可以繼續(xù)做 Checkpoint ，在有限流執(zhí)行中提供更好的容錯保障。

■ Task 結(jié)束后的兩階段提交

我們在部分 Sink 使用上，例如下圖的 Kafka Sink 上，涉及到 Task 需要依靠 Checkpoint 機制，進行二階段提交，從而保證數(shù)據(jù)的 Exactly-once 一致性。

具體可以這樣說：在 Checkpoint 過程中，每個算子只會進行準(zhǔn)備提交的操作。比如數(shù)據(jù)會提交到外部的臨時存儲目錄下，所有任務(wù)都完成這次 Checkpoint 后會收到一個信號，之后才會執(zhí)行正式的 commit，把所有分布式的臨時文件一次性以事務(wù)的方式提交到外部系統(tǒng)。

這種算法在當(dāng)前有限流的情況下，作業(yè)結(jié)束后并不能保證有 Checkpoint，那么最后一部分?jǐn)?shù)據(jù)如何提交？

在 1.14 中，這個問題得到了解決。Task 處理完所有數(shù)據(jù)之后，必須等待 Checkpoint 完成后才可以正式的退出，這是流批一體方面針對有限流任務(wù)結(jié)束的一些改進。

三、checkpoint 機制

1. 現(xiàn)有 Checkpoint 機制痛點

目前 Flink 觸發(fā) Checkpoint 是依靠 barrier 在算子間進行流通，barrier 隨著算子一直往下游進行發(fā)送，當(dāng)算子下游遇到 barrier 的時候就會進行快照操作，然后再把 barrier 往下游繼續(xù)發(fā)送。對于多路的情況我們會把 barrier 進行對齊，把先到 barrier 的這一路數(shù)據(jù)暫時性的 block，等到兩路 barrier 都到了之后再做快照，最后才會去繼續(xù)往下發(fā)送 barrier。

現(xiàn)有的 Checkpoint 機制存在以下問題：

反壓時無法做出 Checkpoint ：在反壓時候 barrier 無法隨著數(shù)據(jù)往下游流動，造成反壓的時候無法做出 Checkpoint。但是其實在發(fā)生反壓情況的時候，我們更加需要去做出對數(shù)據(jù)的 Checkpoint，因為這個時候性能遇到了瓶頸，是更加容易出問題的階段；

Barrier 對齊阻塞數(shù)據(jù)處理 ：阻塞對齊對于性能上存在一定的影響；

恢復(fù)性能受限于 Checkpoint 間隔 ：在做恢復(fù)的時候，延遲受到多大的影響很多時候是取決于 Checkpoint 的間隔，間隔越大，需要 replay 的數(shù)據(jù)就會越多，從而造成中斷的影響也就會越大。但是目前 Checkpoint 間隔受制于持久化操作的時間，所以沒辦法做的很快。

2. Unaligned Checkpoint

針對這些痛點，F(xiàn)link 在最近幾個版本一直在持續(xù)的優(yōu)化，Unaligned Checkpoint 就是其中一個機制。barrier 算子在到達(dá) input buffer 最前面的時候，就會開始觸發(fā) Checkpoint 操作。它會立刻把 barrier 傳到算子的 OutPut Buffer 的最前面，相當(dāng)于它會立刻被下游的算子所讀取到。通過這種方式可以使得 barrier 不受到數(shù)據(jù)阻塞，解決反壓時候無法進行 Checkpoint 的問題。

當(dāng)我們把 barrier 發(fā)下去后，需要做一個短暫的暫停，暫停的時候會把算子的 State 和 input output buffer 中的數(shù)據(jù)進行一個標(biāo)記，以方便后續(xù)隨時準(zhǔn)備上傳。對于多路情況會一直等到另外一路 barrier 到達(dá)之前數(shù)據(jù)，全部進行標(biāo)注。

通過這種方式整個在做 Checkpoint 的時候，也不需要對 barrier 進行對齊，唯一需要做的停頓就是在整個過程中對所有 buffer 和 state 標(biāo)注。這種方式可以很好的解決反壓時無法做出 Checkpoint ，和 Barrier 對齊阻塞數(shù)據(jù)影響性能處理的問題。

3. Generalized Incremental Checkpoint ［2］

Generalized Incremental Checkpoint 主要是用于減少 Checkpoint 間隔，如左圖 1 所示，在 Incremental Checkpoint 當(dāng)中，先讓算子寫入 state 的 changelog。寫完后才把變化真正的數(shù)據(jù)寫入到 StateTable 上。state 的 changelog 不斷向外部進行持久的存儲化。在這個過程中我們其實無需等待整個 StateTable 去做一個持久化操作，我們只需要保證對應(yīng)的 Checkpoint 這一部分的 changelog 能夠持久化完成，就可以開始做下一次 Checkpoint。StateTable 是以一個周期性的方式，獨立的去對外做持續(xù)化的一個過程。

這兩個過程進行拆分后，就有了從之前的需要做全量持久化 （Per Checkpoint） 變成 增量持久化 （Per Checkpoint） + 后臺周期性全量持久化，從而達(dá)到同樣容錯的效果。在這個過程中，每一次 Checkpoint 需要做持久化的數(shù)據(jù)量減少了，從而使得做 Checkpoint 的間隔能夠大幅度減少。

其實在 RocksDB 也是能支持 Incremental Checkpoint 。但是有兩個問題：

第一個問題是 RocksDB 的 Incremental Checkpoint 是依賴它自己本身的一些實現(xiàn)，當(dāng)中會存在一些數(shù)據(jù)壓縮，壓縮所消耗的時間以及壓縮效果具有不確定性，這個是和數(shù)據(jù)是相關(guān)的；

第二個問題是只能針對特定的 StateBackend 來使用，目前在做的 Generalized Incremental Checkpoint 實際上能夠保證的是，它與 StateBackend 是無關(guān)的，從運行時的機制來保證了一個比較穩(wěn)定、更小的 Checkpoint 間隔。

Unaligned Checkpoint 在 Flink 1.13 就已經(jīng)發(fā)布了，在 1.14 版本主要是針對 bug 的修復(fù)和補充，針對 Generalized Incremental Checkpoint，目前社區(qū)還在做最后的沖刺，比較有希望在 1.14 中和大家見面。［2］

［2］ Generalized Incremental Checkpoint 最終在 1.14 中沒有完成。

四、性能與效率

1. 大規(guī)模作業(yè)調(diào)度的優(yōu)化

構(gòu)建 Pipeline Region 的性能提升：所有由 pipline 邊所連接構(gòu)成的子圖 。在 Flink 任務(wù)調(diào)度中需要通過識別 Pipeline Region 來保證由同一個 Pipline 邊所連接的任務(wù)能夠同時進行調(diào)度。否則有可能上游的任務(wù)開始調(diào)度，但是下游的任務(wù)并沒有運行。從而導(dǎo)致上游運行完的數(shù)據(jù)無法給下游的節(jié)點進行消費，可能會造成死鎖的情況

任務(wù)部署階段：每個任務(wù)都要從哪些上游讀取數(shù)據(jù)，這些信息會生成 Result Partition Deployment Descriptor。

這兩個構(gòu)建過程在之前的版本都有 O （n^2） 的時間復(fù)雜度，主要問題需要對于每個下游節(jié)點去遍歷每一個上游節(jié)點的情況。例如去遍歷每一個上游是不是一個 Pipeline 邊連接的關(guān)系，或者去遍歷它的每一個上游生成對應(yīng)的 Result Partition 信息。

目前通過引入 group 概念，假設(shè)已知上下游 2 個任務(wù)的連接方式是 all-to-all，那相當(dāng)于把所有 Pipeline Region 信息或者 Result Partition 信息以 Group 的形式進行組合，這樣只需知道下游對應(yīng)的是上游的哪一個 group，就可以把一個 O （n^2） 的復(fù)雜度優(yōu)化到了 O （n）。我們用 wordcount 任務(wù)做了一下測試，對比優(yōu)化前后的性能：

從表格中可以看到構(gòu)建速度具有大幅度提升，構(gòu)建 Pipeline Region 的性能從秒級提升至毫秒級別。任務(wù)部署我們是從第一個任務(wù)開始部署到所有任務(wù)開始運行的狀態(tài)，這邊只統(tǒng)計了流，因為批需要上游結(jié)束后才能結(jié)束調(diào)度。從整體時間來看，整個任務(wù)初始化，調(diào)度以及部署的階段，大概能夠減少分鐘級的時間消耗。

2. 細(xì)粒度資源管理

細(xì)粒度資源管理在過去很多的版本都一直在做，在 Flink1.14 終于可以把這一部分 API 開放出來在 DataSteam 提供給用戶使用了。用戶可以在 DataStream 中自定義 SlotSharingGroup 的劃分情況，如下圖所示的方式去定義 Slot 的資源劃分，實現(xiàn)了支持 DataStream API，自定義 SSG 劃分方式以及資源配置 TaskManager 動態(tài)資源扣減。

對于每一個 Slot 可以通過比較細(xì)粒度的配置，我們在 Runtime 上會自動根據(jù)用戶資源配置進行動態(tài)的資源切割。

這樣做的好處是不會像之前那樣有固定資源的 Slot，而是做資源的動態(tài)扣減，通過這樣的方式希望能夠達(dá)到更加精細(xì)的資源管理和資源的使用率。

五、Table / SQL / Python API

1. Table API / SQL

Window Table-Valued Function 支持更多算子與窗口類型 ，可以看如下表格的對比：

從表格中可以看出對于原有的三個窗口類型進行加強，同時新增 Session 窗口類型，目前支持 Aggregate 的操作。

■ 1.1 支持聲明式注冊 Source/Sink

Table API 支持使用聲明式的方式注冊 Source / Sink 功能對齊 SQL DDL；

同時支持 FLIP-27 新的 Source 接口；

new Source 替代舊的 connect（） 接口。

■ 1.2 全新代碼生成器

解決了大家在生成代碼超過 Java 最長代碼限制，新的代碼生成器會對代碼進行拆解，徹底解決代碼超長的問題。

■ 1.3 移除 Flink Planner

新版本中，Blink Planner 將成為 Flink Planner 的唯一實現(xiàn)。

2. Python API

在之前的版本中，如果有先后執(zhí)行的兩個 UDF，它的執(zhí)行過程如下圖左方。在 JVM 上面有 Java 的 Operator，先把數(shù)據(jù)發(fā)給 Python 下面的 UDF 去執(zhí)行，執(zhí)行后又發(fā)回給 Java，然后傳送給下游的 Operator，最后再進行一次 Python 的這種跨進程的傳輸去處理，會導(dǎo)致存在很多次冗余的數(shù)據(jù)傳輸。

在 1.14 版本中，改進如右圖，可以把它們連接在一起，只需要一個來回的 Java 和 Python 進行數(shù)據(jù)通信，通過減少傳輸數(shù)據(jù)次數(shù)就能夠達(dá)到比較好的性能上的提升。

3. 支持 LoopBack 模式

在以往本地執(zhí)行實際是在 Python 的進程中去運行客戶端程序，提交 Java 進程啟動一個迷你集群去執(zhí)行 Java 部分代碼。Java 部分代碼也會和生產(chǎn)環(huán)境部分的一樣，去啟動一個新的 Python 進程去執(zhí)行對應(yīng)的 Python UDF，從圖下可以看出新的進程其實在本地調(diào)試中是沒有必要存在的。

所以支持 lookback 模式后可以讓 Java 的 opt 直接把 UDF 運行在之前 Python client 所運行的相同的進程內(nèi)，通過這種方式：

首先是避免了啟動額外進程所帶來的開銷；

最重要的是在本地調(diào)試中，我們可以在同一個進程內(nèi)能夠更好利用一些工具進行 debug，這個是對開發(fā)者體驗上的一個提升。

六、總結(jié)

本文主要講解了 Flink1.14 的主要新特性介紹：

首先介紹了目前社區(qū)在批流一體上的工作，通過介紹批流不同的執(zhí)行模式和 JM 節(jié)點任務(wù)觸發(fā)的優(yōu)化改進更好的去兼容批作業(yè)；

然后通過分析現(xiàn)有的 Checkpoint 機制痛點，在新版本中如何改進，以及在大規(guī)模作業(yè)調(diào)度優(yōu)化和細(xì)粒度的資源管理上面如何做到對性能優(yōu)化；

最后介紹了 TableSQL API 和 Pyhton上相關(guān)的性能優(yōu)化。

責(zé)任編輯：haq