2019阿裡雲峰會·上海開發者大會于7月24日盛大開幕，本次峰會與未來世界的開發者們分享開源大資料、IT基礎設施雲化、資料庫、雲原生、物聯網等領域的技術幹貨，共同探讨前沿科技趨勢。本文整理自開源大資料專場中阿裡巴巴進階技術專家楊克特（魯尼）先生的精彩演講，主要講解了Apache Flink過去和現在的發展情況，同時分享了對Apache Flink未來發展方向的了解。 開源大資料專場PPT下載下傳 以下内容根據演講視訊以及PPT整理而成。
一、Flink的過去
1.Flink 的出現
Apache Flink項目在捐獻給Apache之前，是由柏林工業大學博士生發起的項目，當時的Flink系統還是一個基于流式Runtime的批處理引擎，主要解決的也是批處理的問題。2014年，Flink被捐獻給Apache，并迅速成為Apache 的頂級項目之一。2014年8月份，Apache釋出了第一個Flink版本，Flink 0.6.0，在有了較好的流式引擎支援後，流計算的價值也随之被挖掘和重視；同年12月，Flink釋出了0.7版本，正式推出了DataStream API，這也是目前Flink應用的最廣泛的API。
2.Flink 0.9
State的支援和處理是流計算系統難以回避的存在，早期的流計算系統會将State的維護和管理交給使用者，如Storm和Spark Streaming。這種做法會帶來兩個問題，一方面提高了編寫流計算系統的門檻；另一方面，如果使用者自己維護State，容錯成本和系統提供Exactly Once 語義的成本将會提高。是以，2015年6月釋出的Flink 0.9版本引入了内置State支援，并支援多種State 類型，如ValueState、MapState、ListState 等。
同時為了支援 Exactly Once 的一緻性語義，還需要将本地的 State 組裝成一個全局的 Checkpoint。Flink 0.9中引入的Global Checkpoint機制是基于經典的Chandy-Lamport算法進行的改進。如圖，Flink 會在資料源中定期插入Barrier，架構在看到 Barrier 後會對本地的 State 做一個快照，然後再将 Barrier 往下遊發送。我們可以近似的認為處理 Checkpoint 的Barrier隻會引出一個消息處理的 overhead，這和正常的消息處理量相比幾乎可以忽略不計。在引入 Chandy-Lamport 算法以後，Flink 在保證 Exactly Once 的前提下，提供高吞吐和延遲便不再是一個 tradeoff，可以同時保證高吞吐和低延遲，而其它系統在做類似設計時，往往需要在吞吐和延遲之間做取舍，高一緻性會影響吞吐量，反之在大的吞吐下無法保證一緻性。
3.Flink 1.0的基石
Flink 1.0 版本加入了基于事件時間的計算支援，引入了 Watermark 機制，可以高效的容忍亂序資料和遲到資料。Flink 1.0同時還内置支援了各種各樣的 window，開箱即用的滾動、滑動、會話視窗等，還可以靈活地自定義視窗。再加上 Flink 0.9 中加入的 State API 和高效的 Checkpoint 支援，這一切構成了 Flink 1.0 版本的基石。
二、阿裡巴巴與Flink
2015年之後，阿裡巴巴開始注意到 Flink 計算引擎，并且非常認可 Flink 系統設計理念的先進性，看好其發展前景，是以阿裡巴巴内部開始大量使用 Flink，同時也對 Flink 做了大刀闊斧的改進。
1. 重構分布式架構
在阿裡和社群合作之後，考慮到阿裡内部業務資料龐大、線上壓力非常大，是以第一個大刀闊斧的改進就是重構分布式架構。早期的Flink在各個角色之間沒有清晰的劃分，大部分職責集中在同一角色中，比如作業的排程，資源的申請、Task 的配置設定等内容，并且，這個角色還需要管理叢集裡的所有作業，在作業量非常大的阿裡内部場景，很快就暴露了這樣的瓶頸。在重構分布式架構過程中，阿裡有意識的将排程作業和申請資源的角色進行分離，設定了Job Manager和Resource Manager兩個職責，此後Resource Manager可以完全進行插件化處理，友善對接各種資源排程系統，如YARN和Kubernetes。以對接Kubernetes為例，隻需寫一個插件，所有的作業便可以順暢的營運在整個環境中，大大簡化了流程。同時，這個架構還支援每一個作業使用獨立的 Job Manager 和 Resource Manager，這樣也大大提升了擴充性，一個叢集可以輕松支援成千上萬的作業。
2. 增量 Checkpoint
為了解決數十 TB 量級 State 資料，阿裡在 Flink 中引入了增量 Checkpoint 機制。在早期版本中，Flink 在執行 Checkpoint 的時候，會将每個 Task 本地的 State 資料全量拷貝到可靠存儲上。當 State 的量級上到 TB 之後，每次都備份全量的資料顯然是一個無法接受的方案。增量 Checkpoint 機制也比較容易了解，就是在每一次 Checkpoint 時，不将所有 State 資料都重新整理到可靠的存儲上，而隻将這個 Checkpoint 周期内新增的 State 資料進行備份。而在作業碰到異常重新開機恢複的時候，再使用全量的資料進行恢複。有了這個機制之後，Flink 便可以輕松處理數十 TB 的量級 State 資料。這個問題也是當時制約我們内部機器學習系統的最大因素，解決這一問題之後，Flink 流式應用的範圍變得更加廣泛。
3. 基于 credit 的流控機制
Flink 1.0 版本會在多個 Worker 之間共享一個 TCP channel。如果多個 Operator 在一個Task Manager 中，Operator 之間的網絡連接配接又是 TCP 共享，當其中一個 Operator 産生反壓，就會影響到同一個程序中其它 Operator 的處理效率，導緻運作不穩定。是以在網絡層，阿裡引入了基于信用的流控機制，每個 Operator 不能無限制的往 TCP channel 中發送資料。每個 Operator 有自己的信用，當它向下遊發送資料時需要減信用，當下遊真正消費資料後，這個信用分數才會加回來，上遊才可以繼續往這個虛拟 Channel 中發送資料。Flink 引入精細的流控機制之後，作業的吞吐或延遲都變得更加穩定，不會因為某一個算子的臨時抖動導緻整個作業的不穩定。
4. Streaming SQL
阿裡巴巴集團内部有大量的作業，作為平台維護方，如果使用者作業出現問題，需要第一時間檢視使用者的代碼找出問題。但是使用者代碼數量不一，多則上萬行，少則上百行，使得維護成本非常高。是以阿裡選擇統一的 Streaming SQL 作為開發語言，通過檢視使用者的 SQL 就能夠了解使用者的意圖。選擇 SQL 還有很多其他好處，比如 SQL 會內建一個優化器，讓系統和架構幫助使用者優化作業，提升使用者的執行效率。
這裡需要說明一下 Streaming SQL 的語義，這也是一些剛接觸 Streaming SQL 的使用者的典型問題。簡單來說，Streaming SQL和傳統的批處理 SQL 語義上是一緻的，隻是在執行模式和結果輸出方式上有所不同。比如下圖是一個使用者的分數表，需要做簡單的分數求和，同時計算結果的最後更新時間。在 SQL 語句中，SUM(Score) 計算分數，同時取 MAX（Time），與批處理不同之處在于，流式資料的實時性使 Streaming SQL 在運作時無法一下子看到所有資料，如在 12：01 時，Streaming SQL 會數出一個空記錄，以為這時候系統連一條記錄都沒有看到。随着記錄源源不斷的到來，在12：04時輸出第一次的結果，這是對12：04之前記錄的資料都進行了計算。在12：07時，可以看到目前表中所有的資料，對結果進行一次更新輸出。假設 USER_SCORES 表一開始就存在，那麼批處理運作的結果與流計算最終的結果是一樣的，這也就說明了流批一體的 SQL 語義的一緻性。
5. Flink 在阿裡的服務情況
在 2018 年雙 11，阿裡巴巴服務規模已經超過萬台叢集。單作業已經達到了數十 TB 的狀态資料，所有的作業加起來更是達到了 PB 級。每天需要處理超過十萬億的事件資料。在雙 11 的零點峰值時，資料處理量已經達到了 17 億條每秒。
在過去，Flink 基本上圍繞着 Continuous Processing 和 Streaming Analytics 領域展開，包括 DataStream API 和後來提出的 Streaming SQL。Flink 不僅在 Continuous Processing 和 Streaming Analytics 領域站穩了腳跟，并且成為了目前領域的領先者。
三、Flink的現在
1. Flink 1.9的架構變化
目前 Flink 最新的版本是1.9，Flink 在這個版本上做了較大的架構調整。首先，Flink 之前版本的 Table API 和 SQL API 是建構于兩個底層的 API 之上，即 DataStream API 和 DataSet API。Flink 1.9 經曆了較大的架構調整之後，Table API 和 DataStream API 已成為同級的 API。不同之處在于 DataStream API 提供的是更貼近實體執行計劃的 API，引擎完全基于使用者的描述能執行作業，不會過多的進行優化和幹預。Table API 和 SQL 是關系表達式 API，使用者使用這個 API 描述想要做一件什麼事情，由架構在了解使用者意圖之後，配合優化器翻譯成高效的具體執行圖。這兩套 API 在未來都會同時提供流計算和批處理的支援，在此基礎之上，Flink 會共享統一的 DAG 層和 Stream Operator，Runtime 層則保留了分布式的 Streaming DataFlow。
2. 統一 Operator 抽象
Flink 架構的改動引發了統一 Operator 抽象問題，因為原來的 Operator 抽象隻适用于Flink 的 Streaming 作業，Flink 的 DataSet API 并沒有使用原來的 Operator 抽象。Flink 早期的代碼參考了經典資料庫的方式，所有的算子都是以 pull 的模式執行。如下圖， Filter 算子嘗試找上遊拉取資料，上遊算子 HashJoin 會嘗試往兩端（Build 端和 Probe 端）拉取資料，做 Join。在低延遲和高吞吐要求的情況下，Flink 的 Streaming 作業通過推的方式執行，架構在讀取到資料之後會以 push 的方式推給所有需要的 Operator。為了統一 Operator 抽象，讓 Streaming Operator 也能做到 HashJoin 的操作，阿裡在協定上做了擴充，擴充的語義中算子可以通知架構想要的輸入順序。下圖中，HashJoin 通知 Framework 優先将 Build 端資料推給自己，在 HashJoin 處理完 Build 端，同時建構好 Hashtable 之後，再把Probe端的資料推給 HashJoin。以往開發人員支援流或批處理時很多算子需要寫兩套程式，統一 Operator 抽象之後，算子可以實作複用，幫助開發人員提高開發效率，達到事半功倍的效果。
3. Table API & SQL 1.9新特性
全新的 SQL 類型系統：Table API & SQL 1.9 引入了全新的 SQL 的類型系統。以往的Table 層的類型系統複用了 Runtime 的 TypeInformation，但在實際操作過程當中遇到較多的限制。引入全新的 SQL 類型系統可以更好的對齊 SQL 語義。
DDL初步支援：這個版本中 Flink 還引入了 DDL 的初步支援，使用者可以使用 Create Table 或 Drop Table 等簡單的文法定義表格或删除表。
Table API增強：Table API 原來僅為關系表達式的 API，Table API & SQL 1.9中現在加入了 Map，FlatMap 等更加靈活的 API。
統一的Catalog API：Table API & SQL 1.9 引入了統一的 Catalog API 之後，可以友善的和其它的 Catalog 對接。比如常見的 Hive，可以通過統一的 Catalog API，實作與 Hive.metastore 互動的插件，讓 Flink 可以直接讀取和處理 Hive 中的表。
Blink planner：Table API 增加了 Blink planner 的支援，因為在底層的 Runtime 做了較大的變化後，上層需要 SQL 的 Planner 與底層的 Runtime 進行對接。為了確定原來的 Table API 使用者盡量不受影響，社群完整保留了原來的 Flink Planner。但同時又引入了新的 Blink planner，與新的 Runtime 設計進行對接。
Blink Planner Feature
Blink planner 增加了較多的新功能。首先，Blink planner 對資料結構進行了二進制化、增加了更豐富的内置函數、在聚合時引入了 Minibatch 優化、采取多種解熱點手段來解決聚合過程中碰到的熱點資料等。另外，流計算中的維表關聯的應用非常廣泛，開發者需要對資料流進行資料量次元的擴增，是以 Blink Planner 也支援了維表關聯。TopN 在電商領域應用非常廣泛，通過 Blink Planner 提供的 TopN 功能就可以輕松完成統計成交額排名前幾的商家這樣的功能。在對 TopN 功能進行簡單的擴充之後，Blink Planner 還支援了高效的流式去重。值得一提的是，Blink Planner 已經能夠完整的支援批處理，目前阿裡内部版本已經可以跑通完整的 TPC-H 和 TPC-DS 這樣标準的 Benchmark 測試集。
4. 批處理優化
Flink 在 Runtime 層針對批處理實作了較多的優化。批進行中最經典問題便是錯誤處理的恢複。如下圖，Flink 在拓撲中可以比較靈活的調配每個邊的傳輸類型，在 A 跟 B 之間以網絡直連，B 跟 C 之間插入 Cache 層，在輸出端輸出 Cache 資料，減少 FailOver 傳播的代價。假設在 D 節點發生了錯誤，從 D 節點向上回溯到需要重新計算的範圍，當回溯到 Cache 層時，如果 B1 的結果已經存在于 DFS 裡或者 Cache 到了其它地方，錯誤的回溯則不需要再繼續進行。為了確定一緻性，到 Cache 層之後還需繼續向下回溯一遍，對下遊還未執行或執行一半的作業進行簡單的重新開機，如果沒有 Cache 支援，節點之間都是網絡連接配接，當 D 節點發生錯誤時，錯誤會蔓延到整張圖，而在有 Cache 支援的情況下隻需重新開機其中很小的子圖，可以大大提高 Flink 面對錯誤時的恢複效率。
插件化Shuffle Manager：Flink 1.9 版本增加了 Shuffle 插件，使用者自己可以實作中間的Shuffle 層，通過專門的 Service 接收中間的資料。當然也可以複用基于 Yarn 的 Shuffle Service。
5. 生态
Flink 1.9 版本在生态方面有較大的投入，比如增加了 Hive 的相容性。在引入統一的Catelog API 之後，Flink 已經可以直接讀取 Hive Metastore。使用者可以通過 Flink SQL 處理 Hive 中的資料，同時處理完資料之後 Flink 能夠将資料寫回 Hive 表，寫回的方式可以相容 Hive 的資料格式，若有後續的 Hive 作業，使用者可以在 Hive 表上繼續操作。另外，為了給使用者提供更好的開發體驗，Flink 和 Zeppelin 進行了整合，使用者可以直接在 Notebook 中使用 Flink SQL，也可以使用 Python API 編寫 Flink 的作業。
6. 中文社群
Flink 社群對中文使用者非常重視。Flink 社群官網中已經增加了中文版文檔的支援。另外，社群開通了 Flink 中文使用者郵件清單，使用者訂閱郵件清單後，可以使用中文描述問題，社群中會有非常多的熱心愛好者幫助解答問題。
Flink 在實時計算和流計算領域的領先地位已毋庸置疑，後面對批處理支援将會重點關注。從 Flink 1.9 版本中可以看到，無論是推出更強大的 SQL 執行引擎，還是在 Runtime 層對錯誤恢複更友好的支援，都表明了 Flink 1.9 版本對于批處理的重視程度，而這僅僅是開始。
四、Flink 未來發展方向
1. Micro Services 案例
如下圖，電商系統中有訂單層、訂單交易系統、庫存系統、支付系統和物流系統。首先Micro services 之間以事件方式驅動系統之間的調用。使用者觸發一個訂單，訂單系統收到訂單做計算邏輯，再調用庫存系統，以上操作是典型的事件驅動模型。為了保證性能和穩定性，在不同的 Micro Services 中需要使用 RPC Call，如果使用同步的 RPC Call，則需要解決線程資料量膨脹問題，是以需要在 Micro Services 之間會引入 Async Call。由于每個 Micro Service 的處理能力有限，比如當訂單跟庫存的 RPC 比例是 1:10 比例時，我們不能無限制的向下遊系發送 RPC 調用，是以需要引入一套流控的機制，适當放緩發送的 RPC 的量。但使用者流量難以預測，最佳解決方案是每個 Micro Service 都可以單獨的擴容和縮容。回到訂單系統，當訂單系統壓力較大時，對訂單層做擴容，或者當庫存處于流量低峰時，可以進行服務能力的縮減，所有的系統都需要資料的持久化，而系統背後都離不開 DB 的支援。
總結起來，Micro Service 需要幾點核心要素。第一，事件驅動，第二是系統間的異步傳輸，同時需要具備較好的流控機制，在節點之間和節點内做動态的擴縮容，最後需要有自己的 DB，可以了解為 Micro Service 需要有對 State 的支援，能夠存儲曆史狀态。
不難發現，Micro Service 的需求 Flink 都能夠覆寫。首先，Flink 是以消息為驅動的系統，同時有非常精細的流控機制；因為網絡之間天然的解耦，Flink 的資料傳輸都是異步進行；除此之外，Flink 還可以單獨為每一個算子增加并發或者縮減并發，内置 State 的支援等等。Micro Services 的場景遠遠大于流計算和批處理的場景，相信在不遠的将來 Flink 的社群也會朝這個方向做更多的探索和嘗試，實作對 Event-driven Application 服務場景的支援。