阿裡妹導讀：8月22日，Apache Flink 1.9.0 正式釋出。早在今年1月，阿裡便宣布将内部過去幾年打磨的大資料處理引擎Blink進行開源并向 Apache Flink 貢獻代碼。此次版本在結構上有重大變更，修改代碼達150萬行，接下來，我們一起梳理 Flink 1.9.0 中非常值得關注的重要功能與特性。
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 版本究竟由哪些變更而引發了如此大量的修改，以下将詳細說明。
釋出位址
Apache Flink 1.9.0 已經正式釋出，位址：
https://flink.apache.org/news/2019/08/22/release-1.9.0.html架構更新
基本上，系統如果有非常大的變動，那一定是架構更新帶來的。這次也不例外，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 來編寫算子邏輯，包括使用統一的流式分布式執行引擎。
TableAPI & 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 版本還做了幾個相對比較大的重構和新功能，包括：
- FLIP-37：重構 Table API 類型系統
- FLIP-29：Table 增加面向多行多列操作的 API
- FLINK-10232：初步的 SQL DDL 支援
- FLIP-30：全新的統一的 Catalog API
- 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 則提供了更加靈活的通路手段，也能夠讓使用者完成一些比較黑科技的功能：
- 使用者可以使用這個 API 事先從其他外部系統讀取資料，把它們轉存為 Flink Savepoint 的格式，然後讓 Flink 作業從這個 Savepoint 啟動。這樣一來，就能避免很多冷啟動的問題。
- 使用 Flink 的批處理 API 直接分析State 的資料。State 資料一直以來對使用者是個黑盒，這裡面存儲的資料是對是錯，是否有異常，使用者都無進而知。有了這個 API 之後，使用者就可以像分析其他資料一樣，來對 State 資料進行分析。
- 髒資料訂正。假如有一條髒資料污染了你的 State，使用者還可以使用這個 API 對這樣的問題進行修複和訂正。
- 狀态遷移。當使用者修改了作業邏輯，想複用大部分原來作業的 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 的相容一直是社群工作中一個高優先級的事情，相信後續的版本會有持續的改善。
總結
Flink1.9.0 版本經過大半年的緊張開發，終于順利釋出。在這過程中，Flink 社群不僅迎來了相當多的中國開發者和使用者，還迎來了海量的代碼貢獻，預示着一個良好的開端。未來，無論是功能還是生态，我們會繼續在 Flink 社群加大投入，讓 Flink 在整個中國乃至全世界大規模的使用起來。我們也衷心希望有更多的開發者可以加入我們，加入Flink 社群，一起把 Apache Flink 做的越來越好！
原文釋出時間為：2019-08-23
作者： 楊克特（魯尼）
本文來自雲栖社群合作夥伴“
阿裡技術”，了解相關資訊可以關注“
”。