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

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

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

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

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

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

一 綜述

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

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

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

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

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

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

1 Table & SQL 支持 Change Data Capture（CDC）

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

除了考慮到上面的真實需求，Flink 中定義的“Dynamic Table”概念在流上有兩種模型：append 模式和 update 模式。通過 append 模式把流轉化為“Dynamic Table”在之前的版本中已經支持，因此在 1.11.0 中進一步支持 update 模式也從概念層面完整的實現了“Dynamic Table”。

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

在公開的 CDC 調研報告中，Debezium 和 Canal 是用戶中最流行使用的 CDC 工具，這兩種工具用來同步 changelog 到其它的系統(tǒng)中，如消息隊列。據此，FLIP-105 首先支持了 Debezium 和 Canal 這兩種格式，而且 Kafka source 也已經可以支持解析上述格式并輸出更新事件，在后續(xù)的版本中會進一步支持 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 讀寫關系型數據庫或讀取 changelog 時，必須要手動創(chuàng)建對應的 schema。而且當數據庫中的 schema 發(fā)生變化時，也需要手動更新對應的 Flink 作業(yè)以保持一致和類型匹配，任何不匹配都會造成運行時報錯使作業(yè)失敗。用戶經常抱怨這個看似冗余且繁瑣的流程，體驗極差。

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

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

3 Hive 實時數倉

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

1.11.0 在 Hive 生態(tài)中重點實現了實時數倉方案，改善了端到端流式 ETL 的用戶體驗，達到了批流一體 Hive 數倉的目標。同時在兼容性、性能、易用性方面也進一步進行了加強。

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

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

在 Hive 可用性方面的提升：

• FLIP-123 通過 Hive Dialect 為用戶提供語法兼容，這樣用戶無需在 Flink 和 Hive 的 CLI 之間切換，可以直接遷移 Hive 腳本到 Flink 中執(zhí)行。
• 提供 Hive 相關依賴的內置支持，避免用戶自己下載所需的相關依賴。現在只需要單獨下載一個包，配置 HADOOP_CLASSPATH 就可以運行。
• 在 Hive 性能方面，1.10.0 中已經支持了 ORC（Hive 2+）的向量化讀取，1.11.0 中我們補全了所有版本的 Parquet 和 ORC 向量化支持來提升性能。

3 全新 Source API

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

先簡要回顧下 source 之前的主要問題：

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

FLIP-27 在設計時充分考慮了上述的痛點：

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

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

4 PyFlink 生態(tài)

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

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

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

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

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

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

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

1 支持 application 模式和 Kubernetes 增強

1.11.0 版本前，Flink 主要支持如下兩種模式運行：

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

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

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

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

除了支持 application 模式，Flink 原生 K8s 在 1.11.0 中還完善了很多基礎的功能特性（FLINK-14460），以達到生產可用性的標準。例如 Node Selector、Label、Annotation、Toleration 等。為了更方便的與 Hadoop 集成，也支持根據環(huán)境變量自動掛載 Hadoop 配置的功能。

2 Checkpoint & Savepoint 優(yōu)化

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

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

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

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

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

1）嘗試減少數據鏈路中的 buffer 總量（FLINK-16428），這樣 checkpoint barrier 可以盡快被處理對齊。

• 上游輸出端控制單個 sub partition 堆積 buffer 的最大閾值（backlog），避免負載不均場景下單個鏈路上堆積大量 buffer。
• 在不影響網絡吞吐性能的情況下合理修改上下游默認的 buffer 配置。
• 上下游數據傳輸的基礎協(xié)議進行了調整，允許單個數據鏈路可以配置 0 個獨占 buffer 而不死鎖，這樣總的 buffer 數量和作業(yè)并發(fā)規(guī)模解耦。根據實際需求在吞吐性能和 checkpoint 速度兩者之間權衡，自定義 buffer 配比。

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

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

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

Unaligned checkpoint 在反壓嚴重的場景下可以明顯加速 checkpoint 的完成時間，因為它不再依賴于整體的計算吞吐能力，而和系統(tǒng)的存儲性能更加相關，相當于計算和存儲的解耦。但是它的使用也有一定的局限性，它會增加整體 state 的大小，對存儲 IO 帶來額外的開銷，因此在 IO 已經是瓶頸的場景下就不太適合使用 unaligned checkpoint 機制。

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

四 總結

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

經過 1.11.0 “小版本”的短暫調整，Flink 正在醞釀下一個大版本的 feature，相信一定會有很多重量級的特性登場，讓我們拭目以待！

原文鏈接：https://developer.aliyun.com/article/767711?

版權聲明：本文中所有內容均屬于阿里云開發(fā)者社區(qū)所有，任何媒體、網站或個人未經阿里云開發(fā)者社區(qū)協(xié)議授權不得轉載、鏈接、轉貼或以其他方式復制發(fā)布/發(fā)表。申請授權請郵件developerteam@list.alibaba-inc.com，已獲得阿里云開發(fā)者社區(qū)協(xié)議授權的媒體、網站，在轉載使用時必須注明"稿件來源：阿里云開發(fā)者社區(qū)，原文作者姓名"，違者本社區(qū)將依法追究責任。 如果您發(fā)現本社區(qū)中有涉嫌抄襲的內容，歡迎發(fā)送郵件至：developer2020@service.aliyun.com 進行舉報，并提供相關證據，一經查實，本社區(qū)將立刻刪除涉嫌侵權內容。

總結

以上是生活随笔為你收集整理的Flink 1.11.0 发布，有哪些值得关注的新特性？的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。