Flink 1.11.0 发布，有哪些值得关注的新特性？
在進入深度解讀前，我們先簡單了解下社區發布的一般流程，幫助大家更好的理解和參與 Flink 社區的工作。
首先在每個版本的規劃初期，會從志愿者中選出 1-2 名作為 release manager。1.11.0 版本我作為中國這邊的 release manager，同時還有一名來自 Ververica 的 Piotr Nowojski 作為德國方的 release manager，這在某種程度上也說明中國的開發者和貢獻度在整個社區的占比很重要。
接下來會進行這個版本的 feature kickoff。在一些大的方向上，社區的規劃周期可能比較久，會分階段、分步驟跨越多個版本完成，確保質量。每個版本的側重點也會有所不同，比如前兩個版本側重于批處理的加強，而這個版本更側重于流處理易用性的提升。社區規劃的 feature 列表會在郵件列表中發起討論，以收集更多的用戶/開發者意見和反饋。
一般的開發周期為 2-3 個月時間，提前會明確規劃出大概的 feature freeze 時間，之后進行 release candidate 的發布和測試、以及 bug fix。一般經過幾輪的迭代周期后會正式投票通過一個相對穩定的 candidate 版本，然后基于這個版本正式發布。
Flink 1.11.0 從 3 月初的功能規劃到 7 月初的正式發布，歷經了差不多 4 個月的時間，對 Flink 的生態、易用性、生產可用性、穩定性等方面都進行了增強和改善，下面將一一跟大家分享。
一 綜述
Flink 1.11.0 從 feature 凍結后發布了 4 次 candidate 才最終通過。經統計，一共有 236 個貢獻者參與了這次版本開發，解決了 1474 個 jira 問題，涉及 30 多個 FLIP，提交了 2325 個 commit。
縱觀近五次版本發布，可以看出從 1.9.0 開始 Flink 進入了一個快速發展階段，各個維度指標相比之前都有了幾乎翻倍的提高。也是從 1.9.0 開始阿里巴巴內部的 Blink 項目開始被開源 Flink 整合，到 1.10.0 經過兩個大版本已經全部整合完畢，對 Flink 從生態建設、功能性、性能和生產穩定性上都有了大幅的增強。
Flink 1.11.0 版本的最初定位是重點解決易用性問題，提升用戶業務的生產使用體驗，整體上不做大的架構調整和功能開發，傾向于快速迭代的小版本開發。但是從上面統計的各個指標來看，所謂的“小版本”在各個維度的數據也絲毫不遜色于前兩個大版本，解決問題的數量和參與的貢獻者人數也在持續增加，其中來自中國的貢獻者比例達到 62%。
下面我們會深度剖析 Flink 1.11.0 帶來了哪些讓大家期待已久的特性，從用戶直接使用的 API 層一直到執行引擎層，我們都會選擇一些有代表性的 feature 從不同維度解讀，更完整的 feature 列表請大家關注發布的 release blog。
二 生態完善和易用性提升
這兩個維度在某種程度上是相輔相成的，很難嚴格區分開，生態兼容上的缺失常常造成使用上的不便，提升易用性的過程往往也是不斷完善相關生態的過程。在這方面用戶感知最明顯的應該就是 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，如何在外部系統和 Flink 系統之間編解碼這些更新操作是首要解決的問題。考慮到 source 和 sink 是銜接外部系統的一個橋梁，因此 FLIP-95 在定義全新的 Table source 和 Table sink 接口時解決了這個問題。
在公開的 CDC 調研報告中，Debezium 和 Canal 是用戶中最流行使用的 CDC 工具，這兩種工具用來同步 changelog 到其它的系統中，如消息隊列。據此，FLIP-105 首先支持了 Debezium 和 Canal 這兩種格式，而且 Kafka source 也已經可以支持解析上述格式并輸出更新事件，在后續的版本中會進一步支持 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 時，必須要手動創建對應的 schema。而且當數據庫中的 schema 發生變化時，也需要手動更新對應的 Flink 作業以保持一致和類型匹配，任何不匹配都會造成運行時報錯使作業失敗。用戶經常抱怨這個看似冗余且繁瑣的流程，體驗極差。
實際上對于任何和 Flink 連接的外部系統都可能有類似的上述問題，在 1.11.0 中重點解決了和關系型數據庫對接的這個問題。FLIP-93 提供了 JDBC catalog 的基礎接口以及 Postgres catalog 的實現，這樣方便后續實現與其它類型的關系型數據庫的對接。
1.11.0 版本后，用戶使用 Flink SQL 時可以自動獲取表的 schema 而不再需要輸入 DDL。除此之外，任何 schema 不匹配的錯誤都會在編譯階段提前進行檢查報錯，避免了之前運行時報錯造成的作業失敗。這是提升易用性和用戶體驗的一個典型例子。
3 Hive 實時數倉
從 1.9.0 版本開始 Flink 從生態角度致力于集成 Hive，目標打造批流一體的 Hive 數倉。經過前兩個版本的迭代，已經達到了 batch 兼容且生產可用，在 TPC-DS 10T benchmark 下性能達到 Hive 3.0 的 7 倍以上。
1.11.0 在 Hive 生態中重點實現了實時數倉方案，改善了端到端流式 ETL 的用戶體驗，達到了批流一體 Hive 數倉的目標。同時在兼容性、性能、易用性方面也進一步進行了加強。
在實時數倉的解決方案中，憑借 Flink 的流式處理優勢做到實時讀寫 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，通過監控 partition 生成增量讀取新 partition，或者監控文件夾內新文件生成來增量讀取新文件。
在 Hive 可用性方面的提升：
- FLIP-123 通過 Hive Dialect 為用戶提供語法兼容，這樣用戶無需在 Flink 和 Hive 的 CLI 之間切換，可以直接遷移 Hive 腳本到 Flink 中執行。
- 提供 Hive 相關依賴的內置支持，避免用戶自己下載所需的相關依賴。現在只需要單獨下載一個包，配置 HADOOP_CLASSPATH 就可以運行。
- 在 Hive 性能方面，1.10.0 中已經支持了 ORC（Hive 2+）的向量化讀取，1.11.0 中我們補全了所有版本的 Parquet 和 ORC 向量化支持來提升性能。
3 全新 Source API
前面也提到過，source 和 sink 是 Flink 對接外部系統的一個橋梁，對于完善生態、可用性及端到端的用戶體驗是很重要的環節。社區早在一年前就已經規劃了 source 端的徹底重構，從 FLIP-27 的 ID 就可以看出是很早的一個 feature。但是由于涉及到很多復雜的內部機制和考慮到各種 source connector 的實現，設計上需要考慮的很全面。從 1.10.0 就開始做 POC 的實現，最終趕上了 1.11.0 版本的發布。
先簡要回顧下 source 之前的主要問題：
- 對用戶而言，在 Flink 中改造已有的 source 或者重新實現一個生產級的 source connector 不是一件容易的事情，具體體現在沒有公共的代碼可以復用，而且需要理解很多 Flink 內部細節以及實現具體的 event time 分配、watermark 產出、idleness 監測、線程模型等。
- 批和流的場景需要實現不同的 source。
- partitions/splits/shards 概念在接口中沒有顯式表達，比如 split 的發現邏輯和數據消費都耦合在 source function 的實現中，這樣在實現 Kafka 或 Kinesis 類型的 source 時增加了復雜性。
- 在 runtime 執行層，checkpoint 鎖被 source function 搶占會帶來一系列問題，框架很難進行優化。
FLIP-27 在設計時充分考慮了上述的痛點：
- 首先在 Job Manager 和 Task Manager 中分別引入兩種不同的組件 Split Enumerator 和 Source reader，解耦 split 發現和對應的消費處理，同時方便隨意組合不同的策略。比如現有的 Kafka connector 中有多種不同的 partition 發現策略和實現耦合在一起，在新的架構下，我們只需要實現一種 source reader，就可以適配多種 split enumerator 的實現來對應不同的 partition 發現策略。
- 在新架構下實現的 source connector 可以做到批流統一，唯一的小區別是對批場景的有限輸入，split enumerator 會產出固定數量的 split 集合并且每個 split 都是有限數據集；對于流場景的無限輸入，split enumerator 要么產出無限多的 split 或者 split 自身是無限數據集。
- 復雜的 timestamp assigner 以及 watermark generator 透明的內置在 source reader 模塊內運行，對用戶來說是無感知的。這樣用戶如果想實現新的 source connector，一般不再需要重復實現這部分功能。
目前 Flink 已有的 source connector 會在后續的版本中基于新架構來重新實現，legacy source 也會繼續維護幾個版本保持兼容性，用戶也可以按照 release 文檔中的說明來嘗試體驗新 source 的開發。
4 PyFlink 生態
眾所周知，Python 語言在機器學習和數據分析領域有著廣泛的使用。Flink 從 1.9.0 版本開始發力兼容 Python 生態，Python 和 Flink 合力為 PyFlink，把 Flink 的實時分布式處理能力輸出給 Python 用戶。前兩個版本 PyFlink 已經支持了 Python Table API 和 UDF，在 1.11.0 中擴大對 Python 生態庫 Pandas 的支持以及和 SQL DDL/Client 的集成，同時 Python UDF 性能有了極大的提升。
具體來說，之前普通的 Python UDF 每次調用只能處理一條數據，而且在 Java 端和 Python 端都需要序列化/反序列化，開銷很大。1.11.0 中 Flink 支持在 Table & SQL 作業中自定義和使用向量化 Python UDF，用戶只需要在 UDF 修飾中額外增加一個參數 udf_type=“pandas” 即可。這樣帶來的好處是：
- 每次調用可以處理 N 條數據。
- 數據格式基于 Apache Arrow，大大降低了 Java、Python 進程之間的序列化/反序列化開銷。
- 方便 Python 用戶基于 Numpy 和 Pandas 等數據分析領域常用的 Python 庫，開發高性能的 Python UDF。
除此之外，1.11.0 中 PyFlink 還支持：
- PyFlink table 和 Pandas DataFrame 之間無縫切換（FLIP-120），增強 Pandas 生態的易用性和兼容性。
- Table & SQL 中可以定義和使用 Python UDTF（FLINK-14500），不再必需 Java/Scala UDTF。
- Cython 優化 Python UDF 的性能（FLIP-121），對比 1.10.0 可以提升 30 倍。
- Python UDF 中用戶自定義 metric（FLIP-112），方便監控和調試 UDF 的執行。
上述解讀的都是側重 API 層面，用戶開發作業可以直接感知到的易用性的提升。下面我們看看執行引擎層在 1.11.0 中都有哪些值得關注的變化。
三 生產可用性和穩定性提升
1 支持 application 模式和 Kubernetes 增強
1.11.0 版本前，Flink 主要支持如下兩種模式運行：
- Session 模式：提前啟動一個集群，所有作業都共享這個集群的資源運行。優勢是避免每個作業單獨啟動集群帶來的額外開銷，缺點是隔離性稍差。如果一個作業把某個 Task Manager（TM）容器搞掛，會導致這個容器內的所有作業都跟著重啟。雖然每個作業有自己獨立的 Job Manager（JM）來管理，但是這些 JM 都運行在一個進程中，容易帶來負載上的瓶頸。
- Per-job 模式：為了解決 session 模式隔離性差的問題，每個作業根據資源需求啟動獨立的集群，每個作業的 JM 也是運行在獨立的進程中，負載相對小很多。
以上兩種模式的共同問題是需要在客戶端執行用戶代碼，編譯生成對應的 Job Graph 提交到集群運行。在這個過程需要下載相關 jar 包并上傳到集群，客戶端和網絡負載壓力容易成為瓶頸，尤其當一個客戶端被多個用戶共享使用。
1.11.0 中引入了 application 模式（FLIP-85）來解決上述問題，按照 application 粒度來啟動一個集群，屬于這個 application 的所有 job 在這個集群中運行。核心是 Job Graph 的生成以及作業的提交不在客戶端執行，而是轉移到 JM 端執行，這樣網絡下載上傳的負載也會分散到集群中，不再有上述 client 單點上的瓶頸。
用戶可以通過 bin/flink run-application 來使用 application 模式，目前 Yarn 和 Kubernetes（K8s）都已經支持這種模式。Yarn application 會在客戶端將運行作業需要的依賴都通過 Yarn Local Resource 傳遞到 JM。K8s application 允許用戶構建包含用戶 jar 與依賴的鏡像，同時會根據作業自動創建 TM，并在結束后銷毀整個集群，相比 session 模式具有更好的隔離性。K8s 不再有嚴格意義上的 per-job 模式，application 模式相當于 per-job 在集群進行提交作業的實現。
除了支持 application 模式，Flink 原生 K8s 在 1.11.0 中還完善了很多基礎的功能特性（FLINK-14460），以達到生產可用性的標準。例如 Node Selector、Label、Annotation、Toleration 等。為了更方便的與 Hadoop 集成，也支持根據環境變量自動掛載 Hadoop 配置的功能。
2 Checkpoint & Savepoint 優化
checkpoint 和 savepoint 機制一直是 Flink 保持先進性的核心競爭力之一，社區在這個領域的改動很謹慎，最近的幾個大版本中幾乎沒有大的功能和架構上的調整。在用戶郵件列表中，我們經常能看到用戶反饋和抱怨的相關問題：比如 checkpoint 長時間做不出來失敗，savepoint 在作業重啟后不可用等等。1.11.0 有選擇的解決了一些這方面的常見問題，提高生產可用性和穩定性。
1.11.0 之前， savepoint 中 meta 數據和 state 數據分別保存在兩個不同的目錄中，這樣如果想遷移 state 目錄很難識別這種映射關系，也可能導致目錄被誤刪除，對于目錄清理也同樣有麻煩。1.11.0 把兩部分數據整合到一個目錄下，這樣方便整體轉移和復用。另外，之前 meta 引用 state 采用的是絕對路徑，這樣 state 目錄遷移后路徑發生變化也不可用，1.11.0 把 state 引用改成了相對路徑解決了這個問題（FLINK-5763），這樣 savepoint 的管理維護、復用更加靈活方便。
實際生產環境中，用戶經常遭遇 checkpoint 超時失敗、長時間不能完成帶來的困擾。一旦作業 failover 會造成回放大量的歷史數據，作業長時間沒有進度，端到端的延遲增加。1.11.0 從不同維度對 checkpoint 的優化和提速做了改進，目標實現分鐘甚至秒級的輕量型 checkpoint。
首先，增加了 Checkpoint Coordinator 通知 task 取消 checkpoint 的機制（FLINK-8871），這樣避免 task 端還在執行已經取消的 checkpoint 而對系統帶來不必要的壓力。同時 task 端放棄已經取消的 checkpoint，可以更快的參與執行 coordinator 新觸發的 checkpoint，某種程度上也可以避免新 checkpoint 再次執行超時而失敗。這個優化也對后面默認開啟 local recovery 提供了便利，task 端可以及時清理失效 checkpoint 的資源。
其次，在反壓場景下，整個數據鏈路堆積了大量 buffer，導致 checkpoint barrier 排在數據 buffer 后面，不能被 task 及時處理對齊，也就導致了 checkpoint 長時間不能執行。1.11.0 中從兩個維度對這個問題進行解決：
1）嘗試減少數據鏈路中的 buffer 總量（FLINK-16428），這樣 checkpoint barrier 可以盡快被處理對齊。
- 上游輸出端控制單個 sub partition 堆積 buffer 的最大閾值（backlog），避免負載不均場景下單個鏈路上堆積大量 buffer。
- 在不影響網絡吞吐性能的情況下合理修改上下游默認的 buffer 配置。
- 上下游數據傳輸的基礎協議進行了調整，允許單個數據鏈路可以配置 0 個獨占 buffer 而不死鎖，這樣總的 buffer 數量和作業并發規模解耦。根據實際需求在吞吐性能和 checkpoint 速度兩者之間權衡，自定義 buffer 配比。
這個優化有一部分工作已經在 1.11.0 中完成，剩余部分會在下個版本繼續推進完成。
2）實現了全新的 unaligned checkpoint 機制（FLIP-76）從根本上解決了反壓場景下 checkpoint barrier 對齊的問題。實際上這個想法早在 1.10.0 版本之前就開始醞釀設計，由于涉及到很多模塊的大改動，實現機制和線程模型也很復雜。我們實現了兩種不同方案的原型 POC 進行了測試、性能對比，確定了最終的方案，因此直到 1.11.0 才完成了 MVP 版本，這也是 1.11.0 中執行引擎層唯一的一個重量級 feature。其基本思想可以概括為：
- Checkpoint barrier 跨數據 buffer 傳輸，不在輸入輸出隊列排隊等待處理，這樣就和算子的計算能力解耦，barrier 在節點之間的傳輸只有網絡延時，可以忽略不計。
- 每個算子多個輸入鏈路之間不需要等待 barrier 對齊來執行 checkpoint，第一個到的 barrier 就可以提前觸發 checkpoint，這樣可以進一步提速 checkpoint，不會因為個別鏈路的延遲而影響整體。
- 為了和之前 aligned checkpoint 的語義保持一致，所有未被處理的輸入輸出數據 buffer 都將作為 channel state 在 checkpoint 執行時進行快照持久化，在 failover 時連同 operator state 一同進行恢復。換句話說，aligned 機制保證的是 barrier 前面所有數據必須被處理完，狀態實時體現到 operator state 中；而 unaligned 機制把 barrier 前面的未處理數據所反映的 operator state 延后到 failover restart 時通過 channel state 回放進行體現，從狀態恢復的角度來說最終都是一致的。注意這里雖然引入了額外的 in-flight buffer 的持久化，但是這個過程實際是在 checkpoint 的異步階段完成的，同步階段只是進行了輕量級的 buffer 引用，所以不會過多占用算子的計算時間而影響吞吐性能。
Unaligned checkpoint 在反壓嚴重的場景下可以明顯加速 checkpoint 的完成時間，因為它不再依賴于整體的計算吞吐能力，而和系統的存儲性能更加相關，相當于計算和存儲的解耦。但是它的使用也有一定的局限性，它會增加整體 state 的大小，對存儲 IO 帶來額外的開銷，因此在 IO 已經是瓶頸的場景下就不太適合使用 unaligned checkpoint 機制。
1.11.0 中 unaligned checkpoint 還沒有作為默認模式，需要用戶手動配置來開啟，并且只在 exactly-once 模式下生效。但目前還不支持 savepoint 模式，因為 savepoint 涉及到作業的 rescale 場景，channel state 目前還不支持 state 拆分，在后面的版本會進一步支持，所以 savepoint 目前還是會使用之前的 aligned 模式，在反壓場景下有可能需要很長時間才能完成。
四 總結
Flink 1.11.0 版本的開發過程中，我們看到越來越多來自中國的貢獻者參與到核心功能的開發中，見證了 Flink 在中國的生態發展越來越繁榮，比如來自騰訊公司的貢獻者參與了 K8s、checkpoint 等功能開發，來自字節跳動公司的貢獻者參與了 Table & SQL 層以及引擎網絡層的一些開發。希望更多的公司能夠參與到 Flink 開源社區中，分享在不同領域的經驗，使 Flink 開源技術一直保持先進性，能夠普惠到更多的受眾。
經過 1.11.0 “小版本”的短暫調整，Flink 正在醞釀下一個大版本的 feature，相信一定會有很多重量級的特性登場，讓我們拭目以待！
原文鏈接：https://developer.aliyun.com/article/767711?
版權聲明：本文中所有內容均屬于阿里云開發者社區所有，任何媒體、網站或個人未經阿里云開發者社區協議授權不得轉載、鏈接、轉貼或以其他方式復制發布/發表。申請授權請郵件developerteam@list.alibaba-inc.com，已獲得阿里云開發者社區協議授權的媒體、網站，在轉載使用時必須注明"稿件來源：阿里云開發者社區，原文作者姓名"，違者本社區將依法追究責任。 如果您發現本社區中有涉嫌抄襲的內容，歡迎發送郵件至：developer2020@service.aliyun.com 進行舉報，并提供相關證據，一經查實，本社區將立刻刪除涉嫌侵權內容。總結
以上是生活随笔為你收集整理的Flink 1.11.0 发布，有哪些值得关注的新特性？的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 自研开源框架 Midway Server
- 下一篇： 20大新兴物联网趋势