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

本文主要從兩個層面深度剖析：阿里巴巴對Flink究竟作了哪些優化？

取之開源，用之開源

1、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又大刀闊斧地進行了哪些改進。

2、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 中文社區網站