阿里妹導(dǎo)讀：8月22日,，Apache Flink 1.9.0 正式發(fā)布。早在今年1月,，阿里便宣布將內(nèi)部過去幾年打磨的大數(shù)據(jù)處理引擎Blink進(jìn)行開源并向 Apache Flink 貢獻(xiàn)代碼,。此次版本在結(jié)構(gòu)上有重大變更，修改代碼達(dá)150萬行,，接下來,，我們一起梳理 Flink 1.9.0 中非常值得關(guān)注的重要功能與特性。

Flink 1.9.0是阿里內(nèi)部版本 Blink 合并入 Flink 后的首次發(fā)版,，修改代碼150萬行,，此次發(fā)版不僅在結(jié)構(gòu)上有重大變更，在功能特性上也更加強大與完善,。本文將為大家介紹 Flink 1.9.0 有哪些重大變更與新增功能特性,。

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

• Blink 開源的內(nèi)容主要是阿里巴巴基于開源 Flink 引擎，依托集團(tuán)內(nèi)部業(yè)務(wù),，在流計算和批處理上積累的大量新功能,、性能優(yōu)化、穩(wěn)定性提升等核心代碼,。
• Blink 以分支的形式開源,，即開源后會成為 Apache Flink項目下的一個分支。
• Blink 開源的目標(biāo)不是希望成為另一個活躍的項目,，而是將Flink 做的更好,。通過開源的方式讓大家了解所有 Blink 的實現(xiàn)細(xì)節(jié),，提高 Blink 功能merge進(jìn)入Flink 的效率，與社區(qū)協(xié)作更高效,。

半年的時間過去了,，隨著 Flink 1.9.0 版本的發(fā)布，在此我們可以驕傲的宣布：Blink 團(tuán)隊已經(jīng)實現(xiàn)了之前的諾言,！盡管不是所有功能都順利 merge 回了社區(qū),，但是在我們和社區(qū)的共同努力下，F(xiàn)link 正在朝著它最初的夢想大踏步的邁進(jìn),。

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

• 從解決的 issue 數(shù)量和代碼 commit 數(shù)量來看,，1.9.0 已經(jīng)達(dá)到甚至超過了之前兩個版本的總和,。
• 從修改的代碼行數(shù)來看，達(dá)到了驚人的150 萬行,。雖然受一些模塊重構(gòu)以及 Blink merge 等因素的影響,，但不可否認(rèn)的是，1.9.0 版本一定是 Flink 有史以來開發(fā)者們最活躍的版本,。
• 從Contributor 數(shù)量來看,，F(xiàn)link 也已經(jīng)吸引了越來越多的貢獻(xiàn)者。我相信其中就有不少來自中國的用戶和開發(fā)者,，社區(qū)也響應(yīng)號召開通了中文郵件列表,。

那么，1.9.0 版本究竟由哪些變更而引發(fā)了如此大量的修改,，以下將詳細(xì)說明,。

架構(gòu)升級

基本上，系統(tǒng)如果有非常大的變動,，那一定是架構(gòu)升級帶來的,。這次也不例外，F(xiàn)link 在流批融合的方向上邁進(jìn)了一大步,。首先我們來看一下 Flink之前版本的架構(gòu)圖：

相信熟悉Flink 的讀者們對左邊的架構(gòu)圖一定不會感到陌生。簡單來說,，F(xiàn)link 在其分布式流式執(zhí)行引擎之上,，有兩套相對獨立的 DataStream 和 DataSet API，分別來描述流計算和批處理的作業(yè),。在這兩個 API之上,，則提供了一個流批統(tǒng)一的API，即 Table API 和SQL,。用戶可以使用相同的Table API 程序或者 SQL 來描述流批作業(yè),，只是在運行時需要告訴 Flink 引擎希望以流的形式運行還是以批的流式運行,，此時 Table 層的優(yōu)化器就會將程序優(yōu)化成 DataStream 作業(yè)或者 DataSet 作業(yè)。

但是如果我們仔細(xì)查看 DataStream 和 DataSet 底層的實現(xiàn)細(xì)節(jié),，會發(fā)現(xiàn)這兩個 API 共享的東西其實不多,。它們有各自獨立的翻譯和優(yōu)化的流程，而且在真正運行的時候,，兩者也使用了完全不同的 Task,。這樣的不一致對用戶和開發(fā)者來講可能存在問題。

從用戶的角度來說,，他們在編寫作業(yè)的時候需要在兩個 API 之間進(jìn)行選擇,，而這兩個 API 不僅語義不同，同時支持的 connector 種類也不同,，難免會造成一些困擾,。Table 盡管在 API 上已經(jīng)進(jìn)行了統(tǒng)一，但因為底層實現(xiàn)還是基于 DataStream 和 DataSet,，也會受到剛才不一致的問題的影響,。

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

在 Blink 一些先行探索的基礎(chǔ)之上，我們和社區(qū)的開發(fā)人員進(jìn)行了密切的討論,，最終基本敲定了 Flink 未來的技術(shù)架構(gòu)路線,。

在 Flink 的未來版本中,，我們將舍棄 DataSet API,，用戶的 API 主要會分為偏描述物理執(zhí)行計劃的 DataStream API 以及偏描述關(guān)系型計劃的 Table & SQL。DataStream API 提供給用戶更多的是一種“所見即所得”的體驗,，由用戶自行描述和編排算子的關(guān)系,，引擎不會做過多的干涉和優(yōu)化。而Table API & SQL 則繼續(xù)保持現(xiàn)在的風(fēng)格,，提供關(guān)系表達(dá)式API,，引擎會根據(jù)用戶的意圖來進(jìn)行優(yōu)化,，并選擇最優(yōu)的執(zhí)行計劃。值得一提的是,，以后這兩個 API 都會各自同時提供流計算和批處理的功能,。這兩個用戶 API 之下，在實現(xiàn)層它們都會共享相同的技術(shù)棧,，比如會用統(tǒng)一的 DAG 數(shù)據(jù)結(jié)構(gòu)來描述作業(yè),，使用統(tǒng)一的 StreamOperator 來編寫算子邏輯，包括使用統(tǒng)一的流式分布式執(zhí)行引擎,。

TableAPI & SQL

在開源 Blink 時,，Blink 的Table 模塊已經(jīng)使用了 Flink 未來設(shè)想的新架構(gòu)。因此 Flink 1.9 版本中,，Table 模塊順理成章的成為了架構(gòu)調(diào)整后第一個吃螃蟹的人,。但是，為了盡量不影響之前版本用戶的體驗,，我們還是需要找到一個方式讓兩種架構(gòu)能夠并存,。

基于這個目的，社區(qū)的開發(fā)人員做了一系列的努力,，包括將 Table 模塊進(jìn)行拆分（FLIP-32,，F(xiàn)LIP 即 Flink Improvement Proposals，專門記錄一些對Flink 做較大修改的提議）,，對 Java 和 Scala 的 API 進(jìn)行依賴梳理,，并且提出了 Planner 接口以支持多種不同的 Planner 實現(xiàn)。Planner 將負(fù)責(zé)具體的優(yōu)化和將 Table 作業(yè)翻譯成執(zhí)行圖的工作,，我們可以將原來的實現(xiàn)全部挪至 Flink Planner 中,，然后把對接新架構(gòu)的代碼放在 Blink Planner里。

圖中的 Query Processor 就是 Planner 的實現(xiàn)

這樣的做法一舉兩得,。不僅讓 Table 模塊在經(jīng)過拆分后更加清晰，更重要的是不影響老版本用戶的體驗,。

在 1.9 版本中,，我們已經(jīng)merge 了大部分當(dāng)初從 Blink 開源出來的 SQL功能。這些都是近幾年在阿里內(nèi)部場景經(jīng)過千錘百煉而沉淀出來的新功能和性能上的優(yōu)化,，相信能夠促使Flink 更上一個臺階,！

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

• FLIP-37：重構(gòu) Table API 類型系統(tǒng)
• FLIP-29：Table 增加面向多行多列操作的 API
• FLINK-10232：初步的 SQL DDL 支持
• FLIP-30：全新的統(tǒng)一的 Catalog API
• FLIP-38：Table API 增加 Python 版本

有了這些新功能加持，再經(jīng)過后續(xù)修復(fù)和完善,，F(xiàn)link Table API 和 SQL 在未來將會發(fā)揮越來越重要的作用,。

批處理改進(jìn)

Flink的批處理功能在 1.9 版本有了重大進(jìn)步,，在架構(gòu)調(diào)整后，F(xiàn)link 1.9 加入了好幾項對批處理的功能改進(jìn),。

首當(dāng)其沖的是優(yōu)化批處理的錯誤恢復(fù)代價：FLIP-1（Fine Grained Recovery from Task Failures）,，從這個 FLIP 的編號就可以看出，該優(yōu)化其實很早就已經(jīng)提出,，1.9 版本終于有機會將 FLIP-1 中未完成的功能進(jìn)行了收尾,。在新版本中，如果批處理作業(yè)有錯誤發(fā)生,，那么 Flink 首先會去計算這個錯誤的影響范圍,，即 Failover Region。因為在批處理作業(yè)中,，有些節(jié)點之間可以通過網(wǎng)絡(luò)進(jìn)行Pipeline 的數(shù)據(jù)傳輸,，但其他一些節(jié)點可以通過 Blocking 的方式先把輸出數(shù)據(jù)存下來，然后下游再去讀取存儲的數(shù)據(jù)的方式進(jìn)行數(shù)據(jù)傳輸,。如果算子輸出的數(shù)據(jù)已經(jīng)完整的進(jìn)行了保存,，那么就沒有必要把這個算子拉起重跑，這樣一來就可以把錯誤恢復(fù)控制在一個相對較小的范圍里,。

如果作業(yè)極端一點，在每一個需要Shuffle 的地方都進(jìn)行數(shù)據(jù)落盤,，那么就和 MapReduce 以及 Spark 的行為類似了,。只是 Flink 支持更高級的用法,，你可以自行控制每種 Shuffle 是使用網(wǎng)絡(luò)來直連,，還是通過文件落盤來進(jìn)行。

有了基于文件的Shuffle 之后,，大家很容易就會聯(lián)想到,，是不是可以把這個 Shuffle 的實現(xiàn)變成插件化。沒錯,，社區(qū)也正在朝這個方向進(jìn)行改進(jìn)：FLIP-31（Pluggable Shuffle Service）,。比如，我們可以利用 Yarn 的 Auxliary Service 來作為一種 Shuffle 的實現(xiàn),，我們甚至可以去寫一個分布式服務(wù)來幫助批處理任務(wù)進(jìn)行Shuffle,。最近，F(xiàn)acebook 也分享了一些這方面的工作,，而且在阿里內(nèi)部,，我們已經(jīng)使用這樣的架構(gòu)，支持了單作業(yè)處理數(shù)百TB 量級的規(guī)模,。Flink 具備了這樣的插件機制后,，可以輕松的對接這些更加高效靈活的實現(xiàn),，讓Shuffle 這個批處理的老大難問題得到較好的解決。

流處理改進(jìn)

流計算畢竟還是 Flink 發(fā)跡的主要領(lǐng)域,，在 1.9 版本當(dāng)然也不能忘了在這方面做一些改進(jìn),。這個版本增加了一個非常實用的功能，即FLIP-43（State Processor API）,。Flink 的 State 數(shù)據(jù)的訪問,，以及由 State 數(shù)據(jù)組成的 Savepoint 的訪問一直是社區(qū)用戶呼聲比較高的一個功能。在 1.9 之前的版本,，F(xiàn)link 開發(fā)了 Queryable State,，不過這個功能的使用場景比較有限，使用效果也不太理想,，因此用的人一直不多,。這次的 State Processor API 則提供了更加靈活的訪問手段，也能夠讓用戶完成一些比較黑科技的功能：

1. 用戶可以使用這個 API 事先從其他外部系統(tǒng)讀取數(shù)據(jù),，把它們轉(zhuǎn)存為 Flink Savepoint 的格式,，然后讓 Flink 作業(yè)從這個 Savepoint 啟動。這樣一來,，就能避免很多冷啟動的問題,。
2. 使用 Flink 的批處理 API 直接分析State 的數(shù)據(jù)。State 數(shù)據(jù)一直以來對用戶是個黑盒,，這里面存儲的數(shù)據(jù)是對是錯,，是否有異常，用戶都無從而知,。有了這個 API 之后,，用戶就可以像分析其他數(shù)據(jù)一樣，來對 State 數(shù)據(jù)進(jìn)行分析,。
3. 臟數(shù)據(jù)訂正,。假如有一條臟數(shù)據(jù)污染了你的 State，用戶還可以使用這個 API 對這樣的問題進(jìn)行修復(fù)和訂正,。
4. 狀態(tài)遷移,。當(dāng)用戶修改了作業(yè)邏輯，想復(fù)用大部分原來作業(yè)的 State,，但又希望做一些微調(diào),。那么就可以使用這個 API 來完成相應(yīng)的工作。

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

說到 Savepoint,，這里也提一下社區(qū)完成的另外一個實用功能,，即FLIP-34（Stop with Savepoint）。大家都知道 Flink 會周期性的進(jìn)行 Checkpoint,，并且維護(hù)了一個全局的狀態(tài)快照,。假如我們碰到這種場景：用戶在兩個Checkpoint 周期中間主動暫停了作業(yè)，然后過一會又進(jìn)行重啟,。這樣,，F(xiàn)link 會自動讀取上一次成功保存的全局狀態(tài)快照，并開始計算上一次全局快照之后的數(shù)據(jù),。雖然這么做能保證狀態(tài)數(shù)據(jù)的不多不少,，但是輸出到 Sink 的卻已經(jīng)有重復(fù)數(shù)據(jù)了。有了這個功能之后,，F(xiàn)link 會在暫停作業(yè)的同時做一次全局快照,，并存儲到Savepoint。下次啟動時,，會從這個 Savepoint 啟動作業(yè),，這樣 Sink 就不會收到預(yù)期外的重復(fù)數(shù)據(jù)了。不過,，這個做法并不能解決作業(yè)在運行過程中自動Failover而引起的輸出到 Sink 數(shù)據(jù)重復(fù)問題,。

Hive集成

Hive一直是 Hadoop 生態(tài)中一股不可忽視的重要力量。為了更好的推廣 Flink 的批處理功能,，和 Hive 的集成必不可少,。在 1.9 版本的開發(fā)過程中，我們也很開心迎來了兩位 Apache Hive PMC 來推進(jìn) Flink 和 Hive 的集成工作,。

首先要解決的是使用 Flink 讀取 Hive 數(shù)據(jù)的問題,。通過 FLIP-30 提出的統(tǒng)一的 Catalog API 的幫助，目前 Flink 已經(jīng)完整打通了對 Hive Meta Store 的訪問,。同時,，我們也增加了 Hive 的 Connector,，目前已支持 CSV, Sequence File, Orc, Parquet 等格式,。用戶只需要配置 HMS 的訪問方式，就可以使用 Flink 直接讀取 Hive 的表進(jìn)行操作,。在此基礎(chǔ)之上,，F(xiàn)link 還增加了對 Hive 自定義函數(shù)的兼容，像 UDF,， UDTF和 UDAF,，都可以直接運行在Flink SQL里。

在寫的支持上，目前Flink 還支持的比較簡單,，暫時只能 INSERT INTO 一張新表,。不過和 Hive 的兼容一直是社區(qū)工作中一個高優(yōu)先級的事情，相信后續(xù)的版本會有持續(xù)的改善,。

總結(jié)

Flink1.9.0 版本經(jīng)過大半年的緊張開發(fā),，終于順利發(fā)布。在這過程中,，F(xiàn)link 社區(qū)不僅迎來了相當(dāng)多的中國開發(fā)者和用戶,，還迎來了海量的代碼貢獻(xiàn)，預(yù)示著一個良好的開端,。未來,，無論是功能還是生態(tài)，我們會繼續(xù)在 Flink 社區(qū)加大投入,，讓 Flink 在整個中國乃至全世界大規(guī)模的使用起來,。我們也衷心希望有更多的開發(fā)者可以加入我們，加入Flink 社區(qū),，一起把 Apache Flink 做的越來越好,！