取之開源，用之開源

一、SQL層

為了能夠真正做到用戶根據自己的業務邏輯開發一套代碼，能夠同時運行在多種不同的場景，Flink首先需要給用戶提供一個統一的API。在經過一番調研之後，阿裏巴巴實時計算認為SQL是一個非常適合的選擇。在批處理領域，SQL已經經歷了幾十年的考驗，是公認的經典。在流計算領域，近年來也不斷有流表二象性、流是表的ChangeLog等理論出現。在這些理論基礎之上，阿裏巴巴提出了動態表的概念，使得流計算也可以像批處理一樣使用SQL來描述，並且邏輯等價。這樣一來，用戶就可以使用SQL來描述自己的業務邏輯，相同的查詢語句在執行時可以是一個批處理任務，也可以是一個高吞吐低延遲的流計算任務，甚至是先使用批處理技術進行歷史數據的計算，然後自動的轉成流計算任務處理最新的實時數據。在這種聲明式的API之下，引擎有了更多的選擇和優化空間。接下來，我們將介紹其中幾個比較重要的優化。

首先是對SQL層的技術架構進行升級和替換。調研過Flink或者使用過Flink的開發者應該知道，Flink有兩套基礎的API，一套是DataStream，另一套是DataSet。DataStream API是針對流式處理的用戶提供，DataSet API是針對批處理用戶提供，但是這兩套API的執行路徑是完全不一樣的，甚至需要生成不同的Task去執行。Flink原生的SQL層在經過一系列優化之後，會根據用戶希望是批處理還是流處理的不同選擇，去調用DataSet或者是DataStream API。這就會造成用戶在日常開發和優化中，經常要面臨兩套幾乎完全獨立的技術棧，很多事情可能需要重復的去做兩遍。這樣也會導致在一邊的技術棧上做的優化，另外一邊就享受不到。因此阿裏巴巴在SQL層提出了全新的Quyer Processor，它主要包括一個流和批可以盡量做到復用的優化層（Query Optimizer）以及基於相同接口的算子層（Query Executor）。這樣一來， 80%以上的工作可以做到兩邊復用，比如一些公共的優化規則，基礎數據結構等等。同時，流和批也會各自保留自己一些獨特的優化和算子，以滿足不同的作業行為。

在SQL層的技術架構統一之後，阿裏巴巴開始尋求一種更高效的基礎數據結構，以便讓Blink在SQL層的執行更加高效。在原生Flink SQL中，都統一使用了一種叫Row的數據結構，它完全由JAVA的一些對象構成關系數據庫中的一行。假如現在的一行數據由一個整型，一個浮點型以及一個字符串組成，那麽Row當中就會包含一個JAVA的Integer、Double和String。眾所周知，這些JAVA的對象在堆內有不少的額外開銷，同時在訪問這些數據的過程中也會引入不必要的裝箱拆箱操作。基於這些問題，阿裏巴巴提出了一種全新的數據結構BinaryRow，它和原來的Row一樣也是表示一個關系數據中的一行，但與之不同的是，它完全使用二進制數據來存儲這些數據。在上述例子中，三個不同類型的字段統一由JAVA的byte[]來表示。這會帶來諸多好處：

• 首先在存儲空間上，去掉了很多無謂的額外消耗，使得對象的存儲更為緊湊；

• 其次在和網絡或者狀態存儲打交道的時候，也可以省略掉很多不必要的序列化反序列化開銷；

• 最後在去掉各種不必要的裝箱拆箱操作之後，整個執行代碼對GC也更加友好。

通過引入這樣一個高效的基礎數據結構，整個SQL層的執行效率得到了一倍以上的提升。

在算子的實現層面，阿裏巴巴引入了更廣範圍的代碼生成技術。得益於技術架構和基礎數據結構的統一，很多代碼生成技術得以達到更廣範圍的復用。同時由於SQL的強類型保證，用戶可以預先知道算子需要處理的數據的類型，從而可以生成更有針對性更高效的執行代碼。在原生Flink SQL中，只有類似a > 2或者c + d這樣的簡單表達式才會應用代碼生成技術，在阿裏巴巴優化之後，有一些算子會進行整體的代碼生成，比如排序、聚合等。這使得用戶可以更加靈活的去控制算子的邏輯，也可以直接將最終運行代碼嵌入到類當中，去掉了昂貴的函數調用開銷。一些應用代碼生成技術的基礎數據結構和算法，比如排序算法，基於二進制數據的HashMap等，也可以在流和批的算子之間進行共享和復用，讓用戶真正享受到了技術和架構的統一帶來的好處。在針對批處理的某些場景進行數據結構或者算法的優化之後，流計算的性能也能夠得到提升。接下來，我們聊聊阿裏巴巴在Runtime層對Flink又大刀闊斧地進行了哪些改進。

二、Runtime層

為了讓Flink在Alibaba的大規模生產環境中生根發芽，實時計算團隊如期遇到了各種挑戰，首當其沖的就是如何讓Flink與其他集群管理系統進行整合。Flink原生集群管理模式尚未完善，也無法原生地使用其他其他相對成熟的集群管理系統。基於此，一系列棘手的問題接連浮現：多租戶之間資源如何協調？如何動態的申請和釋放資源？如何指定不同資源類型？

為了解決這個問題，實時計算團隊經歷大量的調研與分析，最終選擇的方案是改造Flink資源調度系統，讓Flink可以原生地跑在Yarn集群之上；並且重構Master架構，讓一個Job對應一個Master，從此Master不再是集群瓶頸。以此為契機，阿裏巴巴和社區聯手推出了全新的Flip-6架構，讓Flink資源管理變成可插拔的架構，為Flink的可持續發展打下了堅實的基礎。如今Flink可以無縫運行在YARN、Mesos和K8s之上，正是這個架構重要性的有力說明。

解決了Flink集群大規模部署問題後，接下來的就是可靠和穩定性，為了保證Flink在生產環境中的高可用，阿裏巴巴著重改善了Flink的FailOver機制。首先是Master的FailOver，Flink原生的Master FailOver會重啟所有的Job，改善後Master任何FailOver都不會影響Job的正常運行；其次引入了Region-based的Task FailOver，盡量減少任何Task的FailOver對用戶造成的影響。有了這些改進的保駕護航，阿裏巴巴的大量業務方開始把實時計算遷移到Flink上運行。

Stateful Streaming是Flink的最大亮點，基於Chandy-Lamport算法的Checkpoint機制讓Flink具備Exactly Once一致性的計算能力，但在早期Flink版本中Checkpoint的性能在大規模數據量下存在一定瓶頸，阿裏巴巴也在Checkpoint上進行了大量改- 進，比如：

• 增量Checkpoint機制：阿裏巴巴生產環境中遇到大JOB有幾十TB State是常事，做一次全量CP地動山搖，成本很高，因此阿裏巴巴研發了增量Checkpoint機制，從此之後CP從暴風驟雨變成了細水長流；
• Checkpoint小文件合並：都是規模惹的禍，隨著整個集群Flink JOB越來越多，CP文件數也水漲船高，最後壓的HDFS NameNode不堪重負，阿裏巴巴
  通過把若幹CP小文件合並成一個大文件的組織方式，最終把NameNode的壓力減少了幾十倍。

雖然說所有的數據可以放在State中，但由於一些歷史的原因，用戶依然有一些數據需要存放在像HBase等一些外部KV存儲中，用戶在Flink Job需要訪問這些外部的數據，但是由於Flink一直都是單線程處理模型，導致訪問外部數據的延遲成為整個系統的瓶頸，顯然異步訪問是解決這個問題的直接手段，但是讓用戶在UDF中寫多線程同時還要保證ExactlyOnce語義，卻並非易事。阿裏巴巴在Flink中提出了AsyncOperator，讓用戶在Flink JOB中寫異步調用和寫“Hello Word”一樣簡單 ，這個讓Flink Job的吞吐有了很大的飛躍。

Flink在設計上是一套批流統一的計算引擎，在使用過快如閃電的流計算之後，批用戶也開始有興趣入住Flink小區。但批計算也帶來了新的挑戰，首先在任務調度方面，阿裏巴巴引入了更加靈活的調度機制，能夠根據任務之間的依賴關系進行更加高效的調度；其次就是數據Shuffle，Flink原生的Shuffle Service和TM綁定，任務執行完之後要依舊保持TM無法釋放資源；還有就是原有的Batch shuffle沒有對文件進行合並，所以基本無法在生產中使用。阿裏巴巴開發了Yarn Shuffle Service功能的同時解決了以上兩個問題。在開發Yarn Shuffle Service的時候，阿裏巴巴發現開發一套新的Shuffle Service非常不便，需要侵入Flink代碼的很多地方，為了讓其他開發者方便的擴展不同Shuffle，阿裏巴巴同時改造了Flink Shuffle架構，讓Flink的Shuffle變成可插拔的架構。目前阿裏巴巴的搜索業務已經在使用Flink Batch Job，並且已經開始服務於生產。

經過3年多打磨，Blink已經在阿裏巴巴開始茁壯生長，但是對Runtime的優化和改進是永無止境的，一大波改進和優化正在路上。

深度剖析阿裏巴巴對Apache Flink的優化與改進