随着人工智能時代的降臨，資料量的爆發，阿裡巴巴的商品資料處理就經常需要面對增量和全量兩套不同的業務流程問題，是以阿裡巴巴就在想：能不能有一套統一的大資料引擎技術，使用者隻需要根據自己的業務邏輯開發一套代碼。這樣在各種不同的場景下，不管是全量資料還是增量資料，亦或者實時處理，一套方案即可全部支援，這就是阿裡巴巴選擇 Flink 的背景和初衷。
彼時的 Flink 不管是規模還是穩定性尚未經曆實踐，成熟度有待商榷。阿裡巴巴實時計算團隊決定在阿裡内部建立一個 Flink 分支 Blink，并對 Flink 進行大量的修改和完善，讓其适應阿裡巴巴這種超大規模的業務場景。那麼，阿裡巴巴對 Flink 究竟做了哪些優化呢？
Apache Flink 概述
Apache Flink（以下簡稱 Flink）是誕生于歐洲的一個大資料研究項目，原名 StratoSphere。該項目是柏林工業大學的一個研究性項目，早期專注于批計算。2014 年，StratoSphere 項目中的核心成員孵化出 Flink，并在同年将 Flink 捐贈 Apache，後來 Flink 順利成為 Apache 的頂級大資料項目。同時 Flink 計算的主流方向被定位為流計算，即用流式計算來做所有大資料的計算工作，這就是 Flink 技術誕生的背景。
2014 年 Flink 作為主攻流計算的大資料引擎開始在開源大資料行業内嶄露頭角。差別于 Storm、Spark Streaming 以及其他流式計算引擎的是：它不僅是一個高吞吐、低延遲的計算引擎，同時還提供很多進階功能。比如它提供有狀态的計算，支援狀态管理，支援強一緻性的資料語義以及支援 Event Time,WaterMark 對消息亂序的處理等。
Flink 的受歡迎還離不開它身上的衆多标簽，其中包括性能優秀（尤其在流計算領域）、高可擴充性、支援容錯，是一種純記憶體式的一個計算引擎，做了記憶體管理方面的大量優化，另外也支援 eventime 的處理、支援超大狀态的 Job（在阿裡巴巴中作業的 state 大小超過 TB 的是非常常見的）、支援 exactly-once 的處理。
阿裡巴巴與 Flink
随着人工智能時代的降臨，資料量的爆發，在典型的大資料的業務場景下資料業務最通用的做法是：選用批處理的技術處理全量資料，采用流式計算處理實時增量資料。在絕大多數的業務場景之下，使用者的業務邏輯在批處理和流處理之中往往是相同的。但是，使用者用于批處理和流處理的兩套計算引擎是不同的。
是以，使用者通常需要寫兩套代碼。毫無疑問，這帶來了一些額外的負擔和成本。阿裡巴巴的商品資料處理就經常需要面對增量和全量兩套不同的業務流程問題，是以阿裡巴巴就在想：能不能有一套統一的大資料引擎技術，使用者隻需要根據自己的業務邏輯開發一套代碼。這樣在各種不同的場景下，不管是全量資料還是增量資料，亦或者實時處理，一套方案即可全部支援，這就是阿裡巴巴選擇 Flink 的背景和初衷。
基于 Flink 在阿裡巴巴搭建的平台于 2016 年正式上線，并從阿裡巴巴的搜尋和推薦這兩大場景開始實作。目前阿裡巴巴所有的業務，包括阿裡巴巴所有子公司都采用了基于 Flink 搭建的實時計算平台。同時 Flink 計算平台運作在開源的 Hadoop 叢集之上。采用 Hadoop 的 YARN 做為資源管理排程，以 HDFS 作為資料存儲。是以，Flink 可以和開源大資料軟體 Hadoop 無縫對接。
目前，這套基于 Flink 搭建的實時計算平台不僅服務于阿裡巴巴集團内部，而且通過阿裡雲的雲産品 API 向整個開發者生态提供基于 Flink 的雲産品支援。
彼時的 Flink 不管是規模還是穩定性尚未經曆實踐，成熟度有待商榷。阿裡巴巴實時計算團隊決定在阿裡内部建立一個 Flink 分支 Blink，并對 Flink 進行大量的修改和完善，讓其适應阿裡巴巴這種超大規模的業務場景。在這個過程當中，該團隊不僅對 Flink 在性能和穩定性上做出了很多改進和優化，同時在核心架構和功能上也進行了大量創新和改進，并将逐漸推回給社群，例如：Flink 新的分布式架構，增量 Checkpoint 機制, 基于 Credit-based 的網絡流控機制和 Streaming SQL 等。接下來，我們主要從兩個層面深度剖析阿裡巴巴對 Flink 究竟做了哪些優化？
取之開源，用之開源一、 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 的優化和改進是永無止境的，一大波改進和優化正在路上。
Flink 的未來方向
目前 Flink 已經成為一個主流的流計算引擎，社群下一步很重要的工作是讓 Flink 在批計算上有所突破，在更多的場景下落地，成為一種主流的批計算引擎。然後進一步在流和批之間進行無縫切換，使流和批的界限越來越模糊。用 Flink，在一個計算中，既可以有流計算，又可以有批計算。
接下來阿裡巴巴還将緻力于推動 Flink 在生态上得到更多語言的支援，不僅僅是 Java、Scala 語言，甚至是機器學習下用的 Python、Go 語言。
另一點不得不說 AI，因為現在很多大資料計算的需求和資料量都是在支援很火爆的 AI 場景，是以 Flink 在流批生态完善的基礎上，将繼續完善上層的 Machine Learning 算法庫，同時 Flink 也會向更成熟的機器學習、深度學習去內建。比如可以做 Tensorflow On Flink, 讓大資料的 ETL 資料處理和機器學習的 Feature 計算和特征計算，訓練的計算等進行內建，讓開發者能夠同時享受到多種生态給大家帶來的好處。
最後，從生态、社群的活躍來說，阿裡巴巴目前在推進的一件事情是籌備 2018 年 12 月 20 日 -21 日在國家會議中心舉辦的首屆 Flink Forward China 峰會（千人規模），參與者将有機會了解阿裡巴巴、騰訊、華為、滴滴、美團、位元組跳動等公司為何将 Flink 作為首選的流處理引擎。
原文釋出時間為：2018-11-5
本文作者：阿裡實時計算團隊
本文來自雲栖社群合作夥伴“
AI前線”，了解相關資訊可以關注“
”。