8月22日，Apache Flink 1.9.0 正式釋出，早在今年1月，阿里便宣佈將內部過去幾年打磨的大資料處理引擎Blink進行開源並向 Apache Flink 貢獻程式碼。當前 Flink 1.9.0是阿里內部版本 Blink 合併入 Flink 後的首次發版，修改程式碼150萬行，此次發版不僅在結構上有重大變更，在功能特性上也更加強大與完善。本文將為大家介紹 Flink 1.9.0 有哪些重大變更與新增功能特性。

在此先簡單回顧一下阿里巴巴 Blink 開源的部分要點：

• Blink 開源的內容主要是阿里巴巴基於開源 Flink 引擎，依託集團內部業務，在流計算和批處理上積累的大量新功能、效能優化、穩定性提升等核心程式碼
• Blink 以分支的形式開源，即開源後會成為 Apache Flink 專案下的一個分支。
• Blink 開源的目標不是希望成為另一個活躍的專案，而是將 Flink 做的更好。通過開源的方式讓大家瞭解所有 Blink 的實現細節，提高 Blink 功能merge進入 Flink 的效率，與社群協作更高效。

半年的時間過去了，隨著 Flink 1.9.0 版本的釋出，在此我們可以驕傲的宣佈：Blink 團隊已經實現了之前的諾言！儘管不是所有功能都順利 merge 回了社群，但是在我們和社群的共同努力下，Flink 正在朝著它最初的夢想大踏步的邁進。

先和大家分享幾個 Flink 1.9.0 版本與之前個版本的對比數字：

• 從解決的 issue 數量和程式碼 commit 數量來看，1.9.0 已經達到甚至超過了之前兩個版本的總和。
• 從修改的程式碼行數來看，達到了驚人的 150 萬行。雖然受一些模組重構以及 Blink merge 等因素的影響，但不可否認的是，1.9.0 版本一定是 Flink 有史以來開發者們最活躍的版本。
• 從 Contributor 數量來看，Flink 也已經吸引了越來越多的貢獻者。我相信其中就有不少來自中國的使用者和開發者，社群也響應號召開通了中文郵件列表。

那麼，1.9.0 版本究竟由哪些變更而引發瞭如此大量的修改，以下將詳細說明。

架構升級

基本上，系統如果有非常大的變動，那一定是架構升級帶來的。這次也不例外，Flink 在流批融合的方向上邁進了一大步。首先我們來看一下 Flink之前版本的架構圖：

相信熟悉 Flink 的讀者們對左邊的架構圖一定不會感到陌生。簡單來說，Flink 在其分散式流式執行引擎之上，有兩套相對獨立的 DataStream 和 DataSet API，分別來描述流計算和批處理的作業。在這兩個 API 之上，則提供了一個流批統一的API，即 Table API 和 SQL。使用者可以使用相同的Table API 程式或者 SQL 來描述流批作業，只是在執行時需要告訴 Flink 引擎希望以流的形式執行還是以批的流式執行，此時 Table 層的優化器就會將程式優化成 DataStream 作業或者 DataSet 作業。

但是如果我們仔細檢視 DataStream 和 DataSet 底層的實現細節，會發現這兩個 API 共享的東西其實不多。它們有各自獨立的翻譯和優化的流程，而且在真正執行的時候，兩者也使用了完全不同的 Task。這樣的不一致對使用者和開發者來講可能存在問題。

從使用者的角度來說，他們在編寫作業的時候需要在兩個 API 之間進行選擇，而這兩個 API 不僅語義不同，同時支援的 connector 種類也不同，難免會造成一些困擾。Table 儘管在 API 上已經進行了統一，但因為底層實現還是基於 DataStream 和 DataSet，也會受到剛才不一致的問題的影響。

從開發者角度來說，由於這兩套流程相對獨立，因此基本上很難做到程式碼的複用。我們在開發一些新功能的時候，往往需要將類似的功能開發兩次，並且每種 API 的開發路徑都比較長，基本都屬於端到端的修改，這大大降低了我們的開發效率。如果兩條獨立的技術棧長期存在，不僅會造成人力的長期浪費，最終可能還會導致整個 Flink 的功能開發變慢。

在 Blink 一些先行探索的基礎之上，我們和社群的開發人員進行了密切的討論，最終基本敲定了 Flink 未來的技術架構路線。

在 Flink 的未來版本中，我們將捨棄 DataSet API，使用者的 API 主要會分為偏描述物理執行計劃的 DataStream API 以及偏描述關係型計劃的 Table & SQL。DataStream API 提供給使用者更多的是一種“所見即所得”的體驗，由使用者自行描述和編排運算元的關係，引擎不會做過多的干涉和優化。而 Table API & SQL 則繼續保持現在的風格，提供關係表示式 API，引擎會根據使用者的意圖來進行優化，並選擇最優的執行計劃。值得一提的是，以後這兩個 API 都會各自同時提供流計算和批處理的功能。這兩個使用者 API 之下，在實現層它們都會共享相同的技術棧，比如會用統一的 DAG 資料結構來描述作業，使用統一的 StreamOperator 來編寫運算元邏輯，包括使用統一的流式分散式執行引擎。

Table API & SQL

在開源 Blink 時，Blink 的 Table 模組已經使用了 Flink 未來設想的新架構。因此 Flink 1.9 版本中，Table 模組順理成章的成為了架構調整後第一個吃螃蟹的人。但是，為了儘量不影響之前版本使用者的體驗，我們還是需要找到一個方式讓兩種架構能夠並存。

基於這個目的，社群的開發人員做了一系列的努力，包括將 Table 模組進行拆分（FLIP-32，FLIP 即 Flink Improvement Proposals，專門記錄一些對 Flink 做較大修改的提議），對 Java 和 Scala 的 API 進行依賴梳理，並且提出了 Planner 介面以支援多種不同的 Planner 實現。Planner 將負責具體的優化和將 Table 作業翻譯成執行圖的工作，我們可以將原來的實現全部挪至 Flink Planner 中，然後把對接新架構的程式碼放在 Blink Planner 裡。

（圖中的 Query Processor 就是 Planner 的實現）

這樣的做法一舉兩得。不僅讓 Table 模組在經過拆分後更加清晰，更重要的是不影響老版本使用者的體驗。

在 1.9 版本中，我們已經merge 了大部分當初從 Blink 開源出來的 SQL功能。這些都是近幾年在阿里內部場景經過千錘百煉而沉澱出來的新功能和效能上的優化，相信能夠促使 Flink 更上一個臺階！

具體功能的相關介紹，請關注社群的釋出申明和使用文件，包括我們後續跟進的系列文章。

除了架構升級之外，Table 模組在 1.9 版本還做了幾個相對比較大的重構和新功能，包括：

1. FLIP-37：重構 Table API 型別系統
2. FLIP-29：Table 增加面向多行多列操作的 API
3. FLINK-10232：初步的 SQL DDL 支援
4. FLIP-30：全新的統一的 Catalog API
5. FLIP-38：Table API 增加 Python 版本

有了這些新功能加持，再經過後續修復和完善，Flink Table API 和 SQL 在未來將會發揮越來越重要的作用。

批處理改進

Flink的批處理功能在 1.9 版本有了重大進步，在架構調整後，Flink 1.9 加入了好幾項對批處理的功能改進。

首當其衝的是優化批處理的錯誤恢復代價：FLIP-1（Fine Grained Recovery from Task Failures），從這個 FLIP 的編號就可以看出，該優化其實很早就已經提出，1.9 版本終於有機會將 FLIP-1 中未完成的功能進行了收尾。在新版本中，如果批處理作業有錯誤發生，那麼 Flink 首先會去計算這個錯誤的影響範圍，即 Failover Region。因為在批處理作業中，有些節點之間可以通過網路進行 Pipeline 的資料傳輸，但其他一些節點可以通過 Blocking 的方式先把輸出資料存下來，然後下游再去讀取儲存的資料的方式進行資料傳輸。如果運算元輸出的資料已經完整的進行了儲存，那麼就沒有必要把這個運算元拉起重跑，這樣一來就可以把錯誤恢復控制在一個相對較小的範圍裡。

如果作業極端一點，在每一個需要 Shuffle 的地方都進行資料落盤，那麼就和 MapReduce 以及 Spark 的行為類似了。只是 Flink 支援更高階的用法，你可以自行控制每種 Shuffle 是使用網路來直連，還是通過檔案落盤來進行。

有了基於檔案的Shuffle 之後，大家很容易就會聯想到，是不是可以把這個 Shuffle 的實現變成外掛化。沒錯，社群也正在朝這個方向進行改進：FLIP-31（Pluggable Shuffle Service）。比如，我們可以利用 Yarn 的 Auxliary Service 來作為一種 Shuffle 的實現，我們甚至可以去寫一個分散式服務來幫助批處理任務進行 Shuffle。最近，Facebook 也分享了一些這方面的工作，而且在阿里內部，我們已經使用這樣的架構，支援了單作業處理數百 TB 量級的規模。Flink 具備了這樣的外掛機制後，可以輕鬆的對接這些更加高效靈活的實現，讓 Shuffle 這個批處理的老大難問題得到較好的解決。

流處理改進

流計算畢竟還是 Flink 發跡的主要領域，在 1.9 版本當然也不能忘了在這方面做一些改進。這個版本增加了一個非常實用的功能，即 FLIP-43（State Processor API）。Flink 的 State 資料的訪問，以及由 State 資料組成的 Savepoint 的訪問一直是社群使用者呼聲比較高的一個功能。在 1.9 之前的版本，Flink 開發了 Queryable State，不過這個功能的使用場景比較有限，使用效果也不太理想，因此用的人一直不多。這次的 State Processor API 則提供了更加靈活的訪問手段，也能夠讓使用者完成一些比較黑科技的功能：

1. 使用者可以使用這個 API 事先從其他外部系統讀取資料，把它們轉存為 Flink Savepoint 的格式，然後讓 Flink 作業從這個 Savepoint 啟動。這樣一來，就能避免很多冷啟動的問題。
2. 使用 Flink 的批處理 API 直接分析 State 的資料。State 資料一直以來對使用者是個黑盒，這裡面儲存的資料是對是錯，是否有異常，使用者都無從而知。有了這個 API 之後，使用者就可以像分析其他資料一樣，來對 State 資料進行分析。
3. 髒資料訂正。假如有一條髒資料汙染了你的 State，使用者還可以使用這個 API 對這樣的問題進行修復和訂正。
4. 狀態遷移。當用戶修改了作業邏輯，想複用大部分原來作業的 State，但又希望做一些微調。那麼就可以使用這個 API 來完成相應的工作。

上面列舉的都是流計算領域非常常見的需求和問題，都有機會通過這個靈活的 API 進行解決，因此我個人非常看好這個 API 的應用前景。

說到 Savepoint，這裡也提一下社群完成的另外一個實用功能，即FLIP-34（Stop with Savepoint）。大家都知道 Flink 會週期性的進行 Checkpoint，並且維護了一個全域性的狀態快照。假如我們碰到這種場景：使用者在兩個 Checkpoint 週期中間主動暫停了作業，然後過一會又進行重啟。這樣，Flink 會自動讀取上一次成功儲存的全域性狀態快照，並開始計算上一次全域性快照之後的資料。雖然這麼做能保證狀態資料的不多不少，但是輸出到 Sink 的卻已經有重複資料了。有了這個功能之後，Flink 會在暫停作業的同時做一次全域性快照，並存儲到 Savepoint。下次啟動時，會從這個 Savepoint 啟動作業，這樣 Sink 就不會收到預期外的重複資料了。不過，這個做法並不能解決作業在執行過程中自動Failover 而引起的輸出到 Sink 資料重複問題。

Hive 整合

Hive 一直是 Hadoop 生態中一股不可忽視的重要力量。為了更好的推廣 Flink 的批處理功能，和 Hive 的整合必不可少。在 1.9 版本的開發過程中，我們也很開心迎來了兩位 Apache Hive PMC 來推進 Flink 和 Hive 的整合工作。

首先要解決的是使用 Flink 讀取 Hive 資料的問題。通過 FLIP-30 提出的統一的 Catalog API 的幫助，目前 Flink 已經完整打通了對 Hive Meta Store 的訪問。同時，我們也增加了 Hive 的 Connector，目前已支援 CSV, Sequence File, Orc, Parquet 等格式。使用者只需要配置 HMS 的訪問方式，就可以使用 Flink 直接讀取 Hive 的表進行操作。在此基礎之上，Flink 還增加了對 Hive 自定義函式的相容，像 UDF， UDTF和 UDAF，都可以直接執行在 Flink SQL裡。

在寫的支援上，目前 Flink 還支援的比較簡單，暫時只能 INSERT INTO 一張新表。不過和 Hive 的相容一直是社群工作中一個高優先順序的事情，相信後續的版本會有持續的改善。

總結

Flink 1.9.0 版本經過大半年的緊張開發，終於順利釋出。在這過程中，Flink 社群不僅迎來了相當多的中國開發者和使用者，還迎來了海量的程式碼貢獻，預示著一個良好的開端。未來，無論是功能還是生態，我們會繼續在 Flink 社群加大投入，讓 Flink 在整個中國乃至全世界大規模的使用起來。我們也衷心希望有更多的開發者可以加入我們，加入 Flink 社群，一起把 Apache Flink 做的越來越好！

原文連結
本文為雲棲社群原創內容，未經