[[333554]]

7 月 7 日，F(xiàn)link 1.11.0 正式發(fā)布。歷時(shí)近 4 個(gè)月，F(xiàn)link 在生態(tài)、易用性、生產(chǎn)可用性、穩(wěn)定性等方面都進(jìn)行了增強(qiáng)和改善。Apache Flink PMC、阿里巴巴高級(jí)技術(shù)專家王治江，同時(shí)也是這個(gè)版本的 release manager 之一，將和大家一一分享，并深度剖析 Flink 1.11.0 帶來了哪些讓大家期待已久的特性，對(duì)一些有代表性的 feature 從不同維度解讀。

在進(jìn)入深度解讀前，我們先簡(jiǎn)單了解下社區(qū)發(fā)布的一般流程，幫助大家更好的理解和參與 Flink 社區(qū)的工作。

首先在每個(gè)版本的規(guī)劃初期，會(huì)從志愿者中選出 1-2 名作為 release manager。1.11.0 版本我作為中國(guó)這邊的 release manager，同時(shí)還有一名來自 Ververica 的 Piotr Nowojski 作為德國(guó)方的 release manager，這在某種程度上也說明中國(guó)的開發(fā)者和貢獻(xiàn)度在整個(gè)社區(qū)的占比很重要。

接下來會(huì)進(jìn)行這個(gè)版本的 feature kickoff。在一些大的方向上，社區(qū)的規(guī)劃周期可能比較久，會(huì)分階段、分步驟跨越多個(gè)版本完成，確保質(zhì)量。每個(gè)版本的側(cè)重點(diǎn)也會(huì)有所不同，比如前兩個(gè)版本側(cè)重于批處理的加強(qiáng)，而這個(gè)版本更側(cè)重于流處理易用性的提升。社區(qū)規(guī)劃的 feature 列表會(huì)在郵件列表中發(fā)起討論，以收集更多的用戶/開發(fā)者意見和反饋。

一般的開發(fā)周期為 2-3 個(gè)月時(shí)間，提前會(huì)明確規(guī)劃出大概的 feature freeze 時(shí)間，之后進(jìn)行 release candidate 的發(fā)布和測(cè)試、以及 bug fix。一般經(jīng)過幾輪的迭代周期后會(huì)正式投票通過一個(gè)相對(duì)穩(wěn)定的 candidate 版本，然后基于這個(gè)版本正式發(fā)布。

Flink 1.11.0 從 3 月初的功能規(guī)劃到 7 月初的正式發(fā)布，歷經(jīng)了差不多 4 個(gè)月的時(shí)間，對(duì) Flink 的生態(tài)、易用性、生產(chǎn)可用性、穩(wěn)定性等方面都進(jìn)行了增強(qiáng)和改善，下面將一一跟大家分享。

一 綜述

Flink 1.11.0 從 feature 凍結(jié)后發(fā)布了 4 次 candidate 才最終通過。經(jīng)統(tǒng)計(jì)，一共有 236 個(gè)貢獻(xiàn)者參與了這次版本開發(fā)，解決了 1474 個(gè) jira 問題，涉及 30 多個(gè) FLIP，提交了 2325 個(gè) commit。

縱觀近五次版本發(fā)布，可以看出從 1.9.0 開始 Flink 進(jìn)入了一個(gè)快速發(fā)展階段，各個(gè)維度指標(biāo)相比之前都有了幾乎翻倍的提高。也是從 1.9.0 開始阿里巴巴內(nèi)部的 Blink 項(xiàng)目開始被開源 Flink 整合，到 1.10.0 經(jīng)過兩個(gè)大版本已經(jīng)全部整合完畢，對(duì) Flink 從生態(tài)建設(shè)、功能性、性能和生產(chǎn)穩(wěn)定性上都有了大幅的增強(qiáng)。

Flink 1.11.0 版本的最初定位是重點(diǎn)解決易用性問題，提升用戶業(yè)務(wù)的生產(chǎn)使用體驗(yàn)，整體上不做大的架構(gòu)調(diào)整和功能開發(fā)，傾向于快速迭代的小版本開發(fā)。但是從上面統(tǒng)計(jì)的各個(gè)指標(biāo)來看，所謂的“小版本”在各個(gè)維度的數(shù)據(jù)也絲毫不遜色于前兩個(gè)大版本，解決問題的數(shù)量和參與的貢獻(xiàn)者人數(shù)也在持續(xù)增加，其中來自中國(guó)的貢獻(xiàn)者比例達(dá)到 62%。

下面我們會(huì)深度剖析 Flink 1.11.0 帶來了哪些讓大家期待已久的特性，從用戶直接使用的 API 層一直到執(zhí)行引擎層，我們都會(huì)選擇一些有代表性的 feature 從不同維度解讀，更完整的 feature 列表請(qǐng)大家關(guān)注發(fā)布的 release blog。

二 生態(tài)完善和易用性提升

這兩個(gè)維度在某種程度上是相輔相成的，很難嚴(yán)格區(qū)分開，生態(tài)兼容上的缺失常常造成使用上的不便，提升易用性的過程往往也是不斷完善相關(guān)生態(tài)的過程。在這方面用戶感知最明顯的應(yīng)該就是 Table & SQL API 層面的使用。

1 Table & SQL 支持 Change Data Capture(CDC)

CDC 被廣泛使用在復(fù)制數(shù)據(jù)、更新緩存、微服務(wù)間同步數(shù)據(jù)、審計(jì)日志等場(chǎng)景，很多公司都在使用開源的 CDC 工具，如 MySQL CDC。通過 Flink 支持在 Table & SQL 中接入和解析 CDC 是一個(gè)強(qiáng)需求，在過往的很多討論中都被提及過，可以幫助用戶以實(shí)時(shí)的方式處理 changelog 流，進(jìn)一步擴(kuò)展 Flink 的應(yīng)用場(chǎng)景，例如把 MySQL 中的數(shù)據(jù)同步到 PG 或 ElasticSearch 中，低延時(shí)的 temporal join 一個(gè) changelog 等。

除了考慮到上面的真實(shí)需求，F(xiàn)link 中定義的“Dynamic Table”概念在流上有兩種模型：append 模式和 update 模式。通過 append 模式把流轉(zhuǎn)化為“Dynamic Table”在之前的版本中已經(jīng)支持，因此在 1.11.0 中進(jìn)一步支持 update 模式也從概念層面完整的實(shí)現(xiàn)了“Dynamic Table”。

為了支持解析和輸出 changelog，如何在外部系統(tǒng)和 Flink 系統(tǒng)之間編解碼這些更新操作是首要解決的問題。考慮到 source 和 sink 是銜接外部系統(tǒng)的一個(gè)橋梁，因此 FLIP-95 在定義全新的 Table source 和 Table sink 接口時(shí)解決了這個(gè)問題。

在公開的 CDC 調(diào)研報(bào)告中，Debezium 和 Canal 是用戶中最流行使用的 CDC 工具，這兩種工具用來同步 changelog 到其它的系統(tǒng)中，如消息隊(duì)列。據(jù)此，F(xiàn)LIP-105 首先支持了 Debezium 和 Canal 這兩種格式，而且 Kafka source 也已經(jīng)可以支持解析上述格式并輸出更新事件，在后續(xù)的版本中會(huì)進(jìn)一步支持 Avro(Debezium) 和 Protobuf(Canal)。

CREATE TABLE my_table (   
...) WITH (   
'connector'='...', -- e.g. 'kafka'   
'format'='debezium-json',   
'debezium-json.schema-include'='true' -- default: false (Debezium can be configured to include or exclude the message schema)   
'debezium-json.ignore-parse-errors'='true' -- default: false 
); 

2 Table & SQL 支持 JDBC Catalog

1.11.0 之前，用戶如果依賴 Flink 的 source/sink 讀寫關(guān)系型數(shù)據(jù)庫(kù)或讀取 changelog 時(shí)，必須要手動(dòng)創(chuàng)建對(duì)應(yīng)的 schema。而且當(dāng)數(shù)據(jù)庫(kù)中的 schema 發(fā)生變化時(shí)，也需要手動(dòng)更新對(duì)應(yīng)的 Flink 作業(yè)以保持一致和類型匹配，任何不匹配都會(huì)造成運(yùn)行時(shí)報(bào)錯(cuò)使作業(yè)失敗。用戶經(jīng)常抱怨這個(gè)看似冗余且繁瑣的流程，體驗(yàn)極差。

實(shí)際上對(duì)于任何和 Flink 連接的外部系統(tǒng)都可能有類似的上述問題，在 1.11.0 中重點(diǎn)解決了和關(guān)系型數(shù)據(jù)庫(kù)對(duì)接的這個(gè)問題。FLIP-93 提供了 JDBC catalog 的基礎(chǔ)接口以及 Postgres catalog 的實(shí)現(xiàn)，這樣方便后續(xù)實(shí)現(xiàn)與其它類型的關(guān)系型數(shù)據(jù)庫(kù)的對(duì)接。

1.11.0 版本后，用戶使用 Flink SQL 時(shí)可以自動(dòng)獲取表的 schema 而不再需要輸入 DDL。除此之外，任何 schema 不匹配的錯(cuò)誤都會(huì)在編譯階段提前進(jìn)行檢查報(bào)錯(cuò)，避免了之前運(yùn)行時(shí)報(bào)錯(cuò)造成的作業(yè)失敗。這是提升易用性和用戶體驗(yàn)的一個(gè)典型例子。

3 Hive 實(shí)時(shí)數(shù)倉(cāng)

從 1.9.0 版本開始 Flink 從生態(tài)角度致力于集成 Hive，目標(biāo)打造批流一體的 Hive 數(shù)倉(cāng)。經(jīng)過前兩個(gè)版本的迭代，已經(jīng)達(dá)到了 batch 兼容且生產(chǎn)可用，在 TPC-DS 10T benchmark 下性能達(dá)到 Hive 3.0 的 7 倍以上。

1.11.0 在 Hive 生態(tài)中重點(diǎn)實(shí)現(xiàn)了實(shí)時(shí)數(shù)倉(cāng)方案，改善了端到端流式 ETL 的用戶體驗(yàn)，達(dá)到了批流一體 Hive 數(shù)倉(cāng)的目標(biāo)。同時(shí)在兼容性、性能、易用性方面也進(jìn)一步進(jìn)行了加強(qiáng)。

在實(shí)時(shí)數(shù)倉(cāng)的解決方案中，憑借 Flink 的流式處理優(yōu)勢(shì)做到實(shí)時(shí)讀寫 Hive：

• Hive 寫入：FLIP-115 完善擴(kuò)展了 FileSystem connector 的基礎(chǔ)能力和實(shí)現(xiàn)，Table/SQL 層的 sink 可以支持各種格式(CSV、Json、Avro、Parquet、ORC)，而且支持 Hive table 的所有格式。
• Partition 支持：數(shù)據(jù)導(dǎo)入 Hive 引入 partition 提交機(jī)制來控制可見性，通過sink.partition-commit.trigger 控制 partition 提交的時(shí)機(jī)，通過 sink.partition-commit.policy.kind 選擇提交策略，支持 SUCCESS 文件和 metastore 提交。
• Hive 讀取：實(shí)時(shí)化的流式讀取 Hive，通過監(jiān)控 partition 生成增量讀取新 partition，或者監(jiān)控文件夾內(nèi)新文件生成來增量讀取新文件。

在 Hive 可用性方面的提升：

• FLIP-123 通過 Hive Dialect 為用戶提供語(yǔ)法兼容，這樣用戶無(wú)需在 Flink 和 Hive 的 CLI 之間切換，可以直接遷移 Hive 腳本到 Flink 中執(zhí)行。
• 提供 Hive 相關(guān)依賴的內(nèi)置支持，避免用戶自己下載所需的相關(guān)依賴?，F(xiàn)在只需要單獨(dú)下載一個(gè)包，配置 HADOOP_CLASSPATH 就可以運(yùn)行。

在 Hive 性能方面，1.10.0 中已經(jīng)支持了 ORC(Hive 2+)的向量化讀取，1.11.0 中我們補(bǔ)全了所有版本的 Parquet 和 ORC 向量化支持來提升性能。

3 全新 Source API

前面也提到過，source 和 sink 是 Flink 對(duì)接外部系統(tǒng)的一個(gè)橋梁，對(duì)于完善生態(tài)、可用性及端到端的用戶體驗(yàn)是很重要的環(huán)節(jié)。社區(qū)早在一年前就已經(jīng)規(guī)劃了 source 端的徹底重構(gòu)，從 FLIP-27 的 ID 就可以看出是很早的一個(gè) feature。但是由于涉及到很多復(fù)雜的內(nèi)部機(jī)制和考慮到各種 source connector 的實(shí)現(xiàn)，設(shè)計(jì)上需要考慮的很全面。從 1.10.0 就開始做 POC 的實(shí)現(xiàn)，最終趕上了 1.11.0 版本的發(fā)布。

先簡(jiǎn)要回顧下 source 之前的主要問題：

• 對(duì)用戶而言，在 Flink 中改造已有的 source 或者重新實(shí)現(xiàn)一個(gè)生產(chǎn)級(jí)的 source connector 不是一件容易的事情，具體體現(xiàn)在沒有公共的代碼可以復(fù)用，而且需要理解很多 Flink 內(nèi)部細(xì)節(jié)以及實(shí)現(xiàn)具體的 event time 分配、watermark 產(chǎn)出、idleness 監(jiān)測(cè)、線程模型等。
• 批和流的場(chǎng)景需要實(shí)現(xiàn)不同的 source。
• partitions/splits/shards 概念在接口中沒有顯式表達(dá)，比如 split 的發(fā)現(xiàn)邏輯和數(shù)據(jù)消費(fèi)都耦合在 source function 的實(shí)現(xiàn)中，這樣在實(shí)現(xiàn) Kafka 或 Kinesis 類型的 source 時(shí)增加了復(fù)雜性。
• 在 runtime 執(zhí)行層，checkpoint 鎖被 source function 搶占會(huì)帶來一系列問題，框架很難進(jìn)行優(yōu)化。

FLIP-27 在設(shè)計(jì)時(shí)充分考慮了上述的痛點(diǎn)：

• 首先在 Job Manager 和 Task Manager 中分別引入兩種不同的組件 Split Enumerator 和 Source reader，解耦 split 發(fā)現(xiàn)和對(duì)應(yīng)的消費(fèi)處理，同時(shí)方便隨意組合不同的策略。比如現(xiàn)有的 Kafka connector 中有多種不同的 partition 發(fā)現(xiàn)策略和實(shí)現(xiàn)耦合在一起，在新的架構(gòu)下，我們只需要實(shí)現(xiàn)一種 source reader，就可以適配多種 split enumerator 的實(shí)現(xiàn)來對(duì)應(yīng)不同的 partition 發(fā)現(xiàn)策略。
• 在新架構(gòu)下實(shí)現(xiàn)的 source connector 可以做到批流統(tǒng)一，唯一的小區(qū)別是對(duì)批場(chǎng)景的有限輸入，split enumerator 會(huì)產(chǎn)出固定數(shù)量的 split 集合并且每個(gè) split 都是有限數(shù)據(jù)集;對(duì)于流場(chǎng)景的無(wú)限輸入，split enumerator 要么產(chǎn)出無(wú)限多的 split 或者 split 自身是無(wú)限數(shù)據(jù)集。
• 復(fù)雜的 timestamp assigner 以及 watermark generator 透明的內(nèi)置在 source reader 模塊內(nèi)運(yùn)行，對(duì)用戶來說是無(wú)感知的。這樣用戶如果想實(shí)現(xiàn)新的 source connector，一般不再需要重復(fù)實(shí)現(xiàn)這部分功能。

目前 Flink 已有的 source connector 會(huì)在后續(xù)的版本中基于新架構(gòu)來重新實(shí)現(xiàn)，legacy source 也會(huì)繼續(xù)維護(hù)幾個(gè)版本保持兼容性，用戶也可以按照 release 文檔中的說明來嘗試體驗(yàn)新 source 的開發(fā)。

4 PyFlink 生態(tài)

眾所周知，Python 語(yǔ)言在機(jī)器學(xué)習(xí)和數(shù)據(jù)分析領(lǐng)域有著廣泛的使用。Flink 從 1.9.0 版本開始發(fā)力兼容 Python 生態(tài)，Python 和 Flink 合力為 PyFlink，把 Flink 的實(shí)時(shí)分布式處理能力輸出給 Python 用戶。前兩個(gè)版本 PyFlink 已經(jīng)支持了 Python Table API 和 UDF，在 1.11.0 中擴(kuò)大對(duì) Python 生態(tài)庫(kù) Pandas 的支持以及和 SQL DDL/Client 的集成，同時(shí) Python UDF 性能有了極大的提升。

具體來說，之前普通的 Python UDF 每次調(diào)用只能處理一條數(shù)據(jù)，而且在 Java 端和 Python 端都需要序列化/反序列化，開銷很大。1.11.0 中 Flink 支持在 Table & SQL 作業(yè)中自定義和使用向量化 Python UDF，用戶只需要在 UDF 修飾中額外增加一個(gè)參數(shù) udf_type=“pandas” 即可。這樣帶來的好處是：

• 每次調(diào)用可以處理 N 條數(shù)據(jù)。
• 數(shù)據(jù)格式基于 Apache Arrow，大大降低了 Java、Python 進(jìn)程之間的序列化/反序列化開銷。
• 方便 Python 用戶基于 Numpy 和 Pandas 等數(shù)據(jù)分析領(lǐng)域常用的 Python 庫(kù)，開發(fā)高性能的 Python UDF。

除此之外，1.11.0 中 PyFlink 還支持：

• PyFlink table 和 Pandas DataFrame 之間無(wú)縫切換(FLIP-120)，增強(qiáng) Pandas 生態(tài)的易用性和兼容性。
• Table & SQL 中可以定義和使用 Python UDTF(FLINK-14500)，不再必需 Java/Scala UDTF。
• Cython 優(yōu)化 Python UDF 的性能(FLIP-121)，對(duì)比 1.10.0 可以提升 30 倍。
• Python UDF 中用戶自定義 metric(FLIP-112)，方便監(jiān)控和調(diào)試 UDF 的執(zhí)行。

上述解讀的都是側(cè)重 API 層面，用戶開發(fā)作業(yè)可以直接感知到的易用性的提升。下面我們看看執(zhí)行引擎層在 1.11.0 中都有哪些值得關(guān)注的變化。

三 生產(chǎn)可用性和穩(wěn)定性提升

1 支持 application 模式和 Kubernetes 增強(qiáng)

1.11.0 版本前，F(xiàn)link 主要支持如下兩種模式運(yùn)行：

• Session 模式：提前啟動(dòng)一個(gè)集群，所有作業(yè)都共享這個(gè)集群的資源運(yùn)行。優(yōu)勢(shì)是避免每個(gè)作業(yè)單獨(dú)啟動(dòng)集群帶來的額外開銷，缺點(diǎn)是隔離性稍差。如果一個(gè)作業(yè)把某個(gè) Task Manager(TM)容器搞掛，會(huì)導(dǎo)致這個(gè)容器內(nèi)的所有作業(yè)都跟著重啟。雖然每個(gè)作業(yè)有自己獨(dú)立的 Job Manager(JM)來管理，但是這些 JM 都運(yùn)行在一個(gè)進(jìn)程中，容易帶來負(fù)載上的瓶頸。
• Per-job 模式：為了解決 session 模式隔離性差的問題，每個(gè)作業(yè)根據(jù)資源需求啟動(dòng)獨(dú)立的集群，每個(gè)作業(yè)的 JM 也是運(yùn)行在獨(dú)立的進(jìn)程中，負(fù)載相對(duì)小很多。

以上兩種模式的共同問題是需要在客戶端執(zhí)行用戶代碼，編譯生成對(duì)應(yīng)的 Job Graph 提交到集群運(yùn)行。在這個(gè)過程需要下載相關(guān) jar 包并上傳到集群，客戶端和網(wǎng)絡(luò)負(fù)載壓力容易成為瓶頸，尤其當(dāng)一個(gè)客戶端被多個(gè)用戶共享使用。

1.11.0 中引入了 application 模式(FLIP-85)來解決上述問題，按照 application 粒度來啟動(dòng)一個(gè)集群，屬于這個(gè) application 的所有 job 在這個(gè)集群中運(yùn)行。核心是 Job Graph 的生成以及作業(yè)的提交不在客戶端執(zhí)行，而是轉(zhuǎn)移到 JM 端執(zhí)行，這樣網(wǎng)絡(luò)下載上傳的負(fù)載也會(huì)分散到集群中，不再有上述 client 單點(diǎn)上的瓶頸。

用戶可以通過 bin/flink run-application 來使用 application 模式，目前 Yarn 和 Kubernetes(K8s)都已經(jīng)支持這種模式。Yarn application 會(huì)在客戶端將運(yùn)行作業(yè)需要的依賴都通過 Yarn Local Resource 傳遞到 JM。K8s application 允許用戶構(gòu)建包含用戶 jar 與依賴的鏡像，同時(shí)會(huì)根據(jù)作業(yè)自動(dòng)創(chuàng)建 TM，并在結(jié)束后銷毀整個(gè)集群，相比 session 模式具有更好的隔離性。K8s 不再有嚴(yán)格意義上的 per-job 模式，application 模式相當(dāng)于 per-job 在集群進(jìn)行提交作業(yè)的實(shí)現(xiàn)。

除了支持 application 模式，F(xiàn)link 原生 K8s 在 1.11.0 中還完善了很多基礎(chǔ)的功能特性(FLINK-14460)，以達(dá)到生產(chǎn)可用性的標(biāo)準(zhǔn)。例如 Node Selector、Label、Annotation、Toleration 等。為了更方便的與 Hadoop 集成，也支持根據(jù)環(huán)境變量自動(dòng)掛載 Hadoop 配置的功能。

2 Checkpoint & Savepoint 優(yōu)化

checkpoint 和 savepoint 機(jī)制一直是 Flink 保持先進(jìn)性的核心競(jìng)爭(zhēng)力之一，社區(qū)在這個(gè)領(lǐng)域的改動(dòng)很謹(jǐn)慎，最近的幾個(gè)大版本中幾乎沒有大的功能和架構(gòu)上的調(diào)整。在用戶郵件列表中，我們經(jīng)常能看到用戶反饋和抱怨的相關(guān)問題：比如 checkpoint 長(zhǎng)時(shí)間做不出來失敗，savepoint 在作業(yè)重啟后不可用等等。1.11.0 有選擇的解決了一些這方面的常見問題，提高生產(chǎn)可用性和穩(wěn)定性。

1.11.0 之前， savepoint 中 meta 數(shù)據(jù)和 state 數(shù)據(jù)分別保存在兩個(gè)不同的目錄中，這樣如果想遷移 state 目錄很難識(shí)別這種映射關(guān)系，也可能導(dǎo)致目錄被誤刪除，對(duì)于目錄清理也同樣有麻煩。1.11.0 把兩部分?jǐn)?shù)據(jù)整合到一個(gè)目錄下，這樣方便整體轉(zhuǎn)移和復(fù)用。另外，之前 meta 引用 state 采用的是絕對(duì)路徑，這樣 state 目錄遷移后路徑發(fā)生變化也不可用，1.11.0 把 state 引用改成了相對(duì)路徑解決了這個(gè)問題(FLINK-5763)，這樣 savepoint 的管理維護(hù)、復(fù)用更加靈活方便。

實(shí)際生產(chǎn)環(huán)境中，用戶經(jīng)常遭遇 checkpoint 超時(shí)失敗、長(zhǎng)時(shí)間不能完成帶來的困擾。一旦作業(yè) failover 會(huì)造成回放大量的歷史數(shù)據(jù)，作業(yè)長(zhǎng)時(shí)間沒有進(jìn)度，端到端的延遲增加。1.11.0 從不同維度對(duì) checkpoint 的優(yōu)化和提速做了改進(jìn)，目標(biāo)實(shí)現(xiàn)分鐘甚至秒級(jí)的輕量型 checkpoint。

首先，增加了 Checkpoint Coordinator 通知 task 取消 checkpoint 的機(jī)制(FLINK-8871)，這樣避免 task 端還在執(zhí)行已經(jīng)取消的 checkpoint 而對(duì)系統(tǒng)帶來不必要的壓力。同時(shí) task 端放棄已經(jīng)取消的 checkpoint，可以更快的參與執(zhí)行 coordinator 新觸發(fā)的 checkpoint，某種程度上也可以避免新 checkpoint 再次執(zhí)行超時(shí)而失敗。這個(gè)優(yōu)化也對(duì)后面默認(rèn)開啟 local recovery 提供了便利，task 端可以及時(shí)清理失效 checkpoint 的資源。

其次，在反壓場(chǎng)景下，整個(gè)數(shù)據(jù)鏈路堆積了大量 buffer，導(dǎo)致 checkpoint barrier 排在數(shù)據(jù) buffer 后面，不能被 task 及時(shí)處理對(duì)齊，也就導(dǎo)致了 checkpoint 長(zhǎng)時(shí)間不能執(zhí)行。1.11.0 中從兩個(gè)維度對(duì)這個(gè)問題進(jìn)行解決：

1)嘗試減少數(shù)據(jù)鏈路中的 buffer 總量(FLINK-16428)，這樣 checkpoint barrier 可以盡快被處理對(duì)齊。

• 上游輸出端控制單個(gè) sub partition 堆積 buffer 的最大閾值(backlog)，避免負(fù)載不均場(chǎng)景下單個(gè)鏈路上堆積大量 buffer。
• 在不影響網(wǎng)絡(luò)吞吐性能的情況下合理修改上下游默認(rèn)的 buffer 配置。
• 上下游數(shù)據(jù)傳輸?shù)幕A(chǔ)協(xié)議進(jìn)行了調(diào)整，允許單個(gè)數(shù)據(jù)鏈路可以配置 0 個(gè)獨(dú)占 buffer 而不死鎖，這樣總的 buffer 數(shù)量和作業(yè)并發(fā)規(guī)模解耦。根據(jù)實(shí)際需求在吞吐性能和 checkpoint 速度兩者之間權(quán)衡，自定義 buffer 配比。

這個(gè)優(yōu)化有一部分工作已經(jīng)在 1.11.0 中完成，剩余部分會(huì)在下個(gè)版本繼續(xù)推進(jìn)完成。

2)實(shí)現(xiàn)了全新的 unaligned checkpoint 機(jī)制(FLIP-76)從根本上解決了反壓場(chǎng)景下 checkpoint barrier 對(duì)齊的問題。實(shí)際上這個(gè)想法早在 1.10.0 版本之前就開始醞釀設(shè)計(jì)，由于涉及到很多模塊的大改動(dòng)，實(shí)現(xiàn)機(jī)制和線程模型也很復(fù)雜。我們實(shí)現(xiàn)了兩種不同方案的原型 POC 進(jìn)行了測(cè)試、性能對(duì)比，確定了最終的方案，因此直到 1.11.0 才完成了 MVP 版本，這也是 1.11.0 中執(zhí)行引擎層唯一的一個(gè)重量級(jí) feature。其基本思想可以概括為：

• Checkpoint barrier 跨數(shù)據(jù) buffer 傳輸，不在輸入輸出隊(duì)列排隊(duì)等待處理，這樣就和算子的計(jì)算能力解耦，barrier 在節(jié)點(diǎn)之間的傳輸只有網(wǎng)絡(luò)延時(shí)，可以忽略不計(jì)。
• 每個(gè)算子多個(gè)輸入鏈路之間不需要等待 barrier 對(duì)齊來執(zhí)行 checkpoint，第一個(gè)到的 barrier 就可以提前觸發(fā) checkpoint，這樣可以進(jìn)一步提速 checkpoint，不會(huì)因?yàn)閭€(gè)別鏈路的延遲而影響整體。
• 為了和之前 aligned checkpoint 的語(yǔ)義保持一致，所有未被處理的輸入輸出數(shù)據(jù) buffer 都將作為 channel state 在 checkpoint 執(zhí)行時(shí)進(jìn)行快照持久化，在 failover 時(shí)連同 operator state 一同進(jìn)行恢復(fù)。換句話說，aligned 機(jī)制保證的是 barrier 前面所有數(shù)據(jù)必須被處理完，狀態(tài)實(shí)時(shí)體現(xiàn)到 operator state 中;而 unaligned 機(jī)制把 barrier 前面的未處理數(shù)據(jù)所反映的 operator state 延后到 failover restart 時(shí)通過 channel state 回放進(jìn)行體現(xiàn)，從狀態(tài)恢復(fù)的角度來說最終都是一致的。注意這里雖然引入了額外的 in-flight buffer 的持久化，但是這個(gè)過程實(shí)際是在 checkpoint 的異步階段完成的，同步階段只是進(jìn)行了輕量級(jí)的 buffer 引用，所以不會(huì)過多占用算子的計(jì)算時(shí)間而影響吞吐性能。

Unaligned checkpoint 在反壓嚴(yán)重的場(chǎng)景下可以明顯加速 checkpoint 的完成時(shí)間，因?yàn)樗辉僖蕾囉谡w的計(jì)算吞吐能力，而和系統(tǒng)的存儲(chǔ)性能更加相關(guān)，相當(dāng)于計(jì)算和存儲(chǔ)的解耦。但是它的使用也有一定的局限性，它會(huì)增加整體 state 的大小，對(duì)存儲(chǔ) IO 帶來額外的開銷，因此在 IO 已經(jīng)是瓶頸的場(chǎng)景下就不太適合使用 unaligned checkpoint 機(jī)制。

1.11.0 中 unaligned checkpoint 還沒有作為默認(rèn)模式，需要用戶手動(dòng)配置來開啟，并且只在 exactly-once 模式下生效。但目前還不支持 savepoint 模式，因?yàn)?savepoint 涉及到作業(yè)的 rescale 場(chǎng)景，channel state 目前還不支持 state 拆分，在后面的版本會(huì)進(jìn)一步支持，所以 savepoint 目前還是會(huì)使用之前的 aligned 模式，在反壓場(chǎng)景下有可能需要很長(zhǎng)時(shí)間才能完成。

四 總結(jié)

Flink 1.11.0 版本的開發(fā)過程中，我們看到越來越多來自中國(guó)的貢獻(xiàn)者參與到核心功能的開發(fā)中，見證了 Flink 在中國(guó)的生態(tài)發(fā)展越來越繁榮，比如來自騰訊公司的貢獻(xiàn)者參與了 K8s、checkpoint 等功能開發(fā)，來自字節(jié)跳動(dòng)公司的貢獻(xiàn)者參與了 Table & SQL 層以及引擎網(wǎng)絡(luò)層的一些開發(fā)。希望更多的公司能夠參與到 Flink 開源社區(qū)中，分享在不同領(lǐng)域的經(jīng)驗(yàn)，使 Flink 開源技術(shù)一直保持先進(jìn)性，能夠普惠到更多的受眾。

經(jīng)過 1.11.0 “小版本”的短暫調(diào)整，F(xiàn)link 正在醞釀下一個(gè)大版本的 feature，相信一定會(huì)有很多重量級(jí)的特性登場(chǎng)，讓我們拭目以待!