一、簡介

1.14 新版本原本規劃有 35 個比較重要的新特性以及優化工作，目前已經有 26 個工作完成；5 個任務不確定是否能準時完成；另外 4 個特性由于時間或者本身設計上的原因，會放到后續版本完成。［1］

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

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

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 版本開始就受到持續的關注，它作為社區 RoadMap 的重要組成部分，是大數據實時化必然的趨勢。但是另一方面，傳統離線的計算需求其實并不會被實時任務完全取代，而是會長期存在。

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

需要維護兩套系統，相應的就需要兩組開發人員，人力的投入成本很高；

另外，兩套數據鏈路處理相似內容帶來維護的風險性和冗余；

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

在這樣的背景下，Flink 社區認定了實時離線一體化的技術路線是比較重要的技術趨勢和方向。

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

下圖是不同的執行模式：

對于無限的數據流，統一采用了流的執行模式。流的執行模式指的是所有計算節點是通過 Pipeline 模式去連接的，Pipeline 是指上游和下游計算任務是同時運行的，隨著上游不斷產出數據，下游同時在不斷消費數據。這種全 Pipeline 的執行方式可以：

通過 eventTime 表示數據是什么時候產生的；

通過 watermark 得知在哪個時間點，數據已經到達了；

通過 state 來維護計算中間狀態；

通過 Checkpoint 做容錯的處理。

這兩種各有優劣，可以根據作業的具體場景來進行選擇。

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

流批的執行模式是有區別的，最主要的就是批的執行模式會有落盤的中間過程，只有當前面任務執行完成，下游的任務才會觸發，這個容錯機制是通過 shuffle 進行容錯的。

這 2 者也各有各的執行優勢：

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

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

Flink 1.14 的優化點主要是針對在流的執行模式下，如何去處理有限數據集。之前處理無限數據集，和現在處理有限數據集最大的區別在于引入了 “任務可能會結束” 的概念。這種情況下帶來了一些新的問題，如下圖：

■ 在流的執行模式下的 Checkpoint 機制

對于無限流，它的 Checkpoint 是由所有的 source 節點進行觸發的，由 source 節點發送 Checkpoint Barrier ，當 Checkpoint Barrier 流過整個作業時候，同時會存儲當前作業所有的 state 狀態。

而在有限流的 Checkpoint 機制中，Task 是有可能提早結束的。上游的 Task 有可能先處理完任務提早退出了，但下游的 Task 卻還在執行中。在同一個 stage 不同并發下，有可能因為數據量不一致導致部分任務提早完成了。這種情況下，在后續的執行作業中，如何進行 Checkpoint？

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

■ Task 結束后的兩階段提交

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

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

這種算法在當前有限流的情況下，作業結束后并不能保證有 Checkpoint，那么最后一部分數據如何提交？

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

三、checkpoint 機制

1. 現有 Checkpoint 機制痛點

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

現有的 Checkpoint 機制存在以下問題：

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

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

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

2. Unaligned Checkpoint

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

當我們把 barrier 發下去后，需要做一個短暫的暫停，暫停的時候會把算子的 State 和 input output buffer 中的數據進行一個標記，以方便后續隨時準備上傳。對于多路情況會一直等到另外一路 barrier 到達之前數據，全部進行標注。

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

3. Generalized Incremental Checkpoint ［2］

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

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

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

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

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

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

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

四、性能與效率

1. 大規模作業調度的優化

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

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

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

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

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

2. 細粒度資源管理

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

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

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

五、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 的唯一實現。

2. Python API

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

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

3. 支持 LoopBack 模式

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

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

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

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

六、總結

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

首先介紹了目前社區在批流一體上的工作，通過介紹批流不同的執行模式和 JM 節點任務觸發的優化改進更好的去兼容批作業；

然后通過分析現有的 Checkpoint 機制痛點，在新版本中如何改進，以及在大規模作業調度優化和細粒度的資源管理上面如何做到對性能優化；

最后介紹了 TableSQL API 和 Pyhton上相關的性能優化。

責任編輯：haq