摘要：基于在阿里巴巴搭建的平臺于年正式上線，并從阿里巴巴的搜索和推薦這兩大場景開始實(shí)現(xiàn)。在經(jīng)過一番調(diào)研之后，阿里巴巴實(shí)時計算認(rèn)為是一個非常適合的選擇。接下來，我們聊聊阿里巴巴在層對又大刀闊斧地進(jìn)行了哪些改進(jìn)。

Apache Flink（以下簡稱Flink）是誕生于歐洲的一個大數(shù)據(jù)研究項(xiàng)目，原名StratoSphere。該項(xiàng)目是柏林工業(yè)大學(xué)的一個研究性項(xiàng)目，早期專注于批計算。2014年，StratoSphere項(xiàng)目中的核心成員孵化出Flink，并在同年將Flink捐贈Apache，后來Flink順利成為Apache的頂級大數(shù)據(jù)項(xiàng)目。同時Flink計算的主流方向被定位為流計算，即用流式計算來做所有大數(shù)據(jù)的計算工作，這就是Flink技術(shù)誕生的背景。

2014年Flink作為主攻流計算的大數(shù)據(jù)引擎開始在開源大數(shù)據(jù)行業(yè)內(nèi)嶄露頭角。區(qū)別于Storm、Spark Streaming以及其他流式計算引擎的是：它不僅是一個高吞吐、低延遲的計算引擎，同時還提供很多高級功能。比如它提供有狀態(tài)的計算，支持狀態(tài)管理，支持強(qiáng)一致性的數(shù)據(jù)語義以及支持Event Time,WaterMark對消息亂序的處理等。

Flink的受歡迎還離不開它身上的眾多標(biāo)簽，其中包括性能優(yōu)秀（尤其在流計算領(lǐng)域）、高可擴(kuò)展性、支持容錯，是一種純內(nèi)存式的一個計算引擎，做了內(nèi)存管理方面的大量優(yōu)化，另外也支持eventime的處理、支持超大狀態(tài)的Job（在阿里巴巴中作業(yè)的state大小超過TB的是非常常見的）、支持exactly-once的處理。

隨著人工智能時代的降臨，數(shù)據(jù)量的爆發(fā)，在典型的大數(shù)據(jù)的業(yè)務(wù)場景下數(shù)據(jù)業(yè)務(wù)最通用的做法是：選用批處理的技術(shù)處理全量數(shù)據(jù)，采用流式計算處理實(shí)時增量數(shù)據(jù)。在絕大多數(shù)的業(yè)務(wù)場景之下，用戶的業(yè)務(wù)邏輯在批處理和流處理之中往往是相同的。但是，用戶用于批處理和流處理的兩套計算引擎是不同的。

因此，用戶通常需要寫兩套代碼。毫無疑問，這帶來了一些額外的負(fù)擔(dān)和成本。阿里巴巴的商品數(shù)據(jù)處理就經(jīng)常需要面對增量和全量兩套不同的業(yè)務(wù)流程問題，所以阿里巴巴就在想：能不能有一套統(tǒng)一的大數(shù)據(jù)引擎技術(shù)，用戶只需要根據(jù)自己的業(yè)務(wù)邏輯開發(fā)一套代碼。這樣在各種不同的場景下，不管是全量數(shù)據(jù)還是增量數(shù)據(jù)，亦或者實(shí)時處理，一套方案即可全部支持，這就是阿里巴巴選擇Flink的背景和初衷。

基于Flink在阿里巴巴搭建的平臺于2016年正式上線，并從阿里巴巴的搜索和推薦這兩大場景開始實(shí)現(xiàn)。目前阿里巴巴所有的業(yè)務(wù)，包括阿里巴巴所有子公司都采用了基于Flink搭建的實(shí)時計算平臺。同時Flink計算平臺運(yùn)行在開源的Hadoop集群之上。采用Hadoop的YARN做為資源管理調(diào)度，以 HDFS作為數(shù)據(jù)存儲。因此，F(xiàn)link可以和開源大數(shù)據(jù)軟件Hadoop無縫對接。

目前，這套基于Flink搭建的實(shí)時計算平臺不僅服務(wù)于阿里巴巴集團(tuán)內(nèi)部，而且通過阿里云的云產(chǎn)品API向整個開發(fā)者生態(tài)提供基于Flink的云產(chǎn)品支持。

彼時的Flink不管是規(guī)模還是穩(wěn)定性尚未經(jīng)歷實(shí)踐，成熟度有待商榷。阿里巴巴實(shí)時計算團(tuán)隊(duì)決定在阿里內(nèi)部建立一個Flink分支Blink，并對Flink進(jìn)行大量的修改和完善，讓其適應(yīng)阿里巴巴這種超大規(guī)模的業(yè)務(wù)場景。在這個過程當(dāng)中，該團(tuán)隊(duì)不僅對Flink在性能和穩(wěn)定性上做出了很多改進(jìn)和優(yōu)化，同時在核心架構(gòu)和功能上也進(jìn)行了大量創(chuàng)新和改進(jìn)，并將逐漸推回給社區(qū)，例如：Flink新的分布式架構(gòu)，增量Checkpoint機(jī)制,基于Credit-based的網(wǎng)絡(luò)流控機(jī)制和Streaming SQL等。接下來，我們主要從兩個層面深度剖析阿里巴巴對Flink究竟做了哪些優(yōu)化？

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

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

在SQL層的技術(shù)架構(gòu)統(tǒng)一之后，阿里巴巴開始尋求一種更高效的基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)，以便讓Blink在SQL層的執(zhí)行更加高效。在原生Flink SQL中，都統(tǒng)一使用了一種叫Row的數(shù)據(jù)結(jié)構(gòu)，它完全由JAVA的一些對象構(gòu)成關(guān)系數(shù)據(jù)庫中的一行。假如現(xiàn)在的一行數(shù)據(jù)由一個整型，一個浮點(diǎn)型以及一個字符串組成，那么Row當(dāng)中就會包含一個JAVA的Integer、Double和String。眾所周知，這些JAVA的對象在堆內(nèi)有不少的額外開銷，同時在訪問這些數(shù)據(jù)的過程中也會引入不必要的裝箱拆箱操作。基于這些問題，阿里巴巴提出了一種全新的數(shù)據(jù)結(jié)構(gòu)BinaryRow，它和原來的Row一樣也是表示一個關(guān)系數(shù)據(jù)中的一行，但與之不同的是，它完全使用二進(jìn)制數(shù)據(jù)來存儲這些數(shù)據(jù)。在上述例子中，三個不同類型的字段統(tǒng)一由JAVA的byte[]來表示。這會帶來諸多好處：

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

其次在和網(wǎng)絡(luò)或者狀態(tài)存儲打交道的時候，也可以省略掉很多不必要的序列化反序列化開銷；

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

通過引入這樣一個高效的基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)，整個SQL層的執(zhí)行效率得到了一倍以上的提升。

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

為了讓Flink在Alibaba的大規(guī)模生產(chǎn)環(huán)境中生根發(fā)芽，實(shí)時計算團(tuán)隊(duì)如期遇到了各種挑戰(zhàn)，首當(dāng)其沖的就是如何讓Flink與其他集群管理系統(tǒng)進(jìn)行整合。Flink原生集群管理模式尚未完善，也無法原生地使用其他其他相對成熟的集群管理系統(tǒng)。基于此，一系列棘手的問題接連浮現(xiàn)：多租戶之間資源如何協(xié)調(diào)？如何動態(tài)的申請和釋放資源？如何指定不同資源類型？

為了解決這個問題，實(shí)時計算團(tuán)隊(duì)經(jīng)歷大量的調(diào)研與分析，最終選擇的方案是改造Flink資源調(diào)度系統(tǒng)，讓Flink可以原生地跑在Yarn集群之上；并且重構(gòu)Master架構(gòu)，讓一個Job對應(yīng)一個Master，從此Master不再是集群瓶頸。以此為契機(jī)，阿里巴巴和社區(qū)聯(lián)手推出了全新的Flip-6架構(gòu)，讓Flink資源管理變成可插拔的架構(gòu)，為Flink的可持續(xù)發(fā)展打下了堅實(shí)的基礎(chǔ)。如今Flink可以無縫運(yùn)行在YARN、Mesos和K8s之上，正是這個架構(gòu)重要性的有力說明。

解決了Flink集群大規(guī)模部署問題后，接下來的就是可靠和穩(wěn)定性，為了保證Flink在生產(chǎn)環(huán)境中的高可用，阿里巴巴著重改善了Flink的FailOver機(jī)制。首先是Master的FailOver，F(xiàn)link原生的Master FailOver會重啟所有的Job，改善后Master任何FailOver都不會影響Job的正常運(yùn)行；其次引入了Region-based的Task FailOver，盡量減少任何Task的FailOver對用戶造成的影響。有了這些改進(jìn)的保駕護(hù)航，阿里巴巴的大量業(yè)務(wù)方開始把實(shí)時計算遷移到Flink上運(yùn)行。

Stateful Streaming是Flink的最大亮點(diǎn)，基于Chandy-Lamport算法的Checkpoint機(jī)制讓Flink具備Exactly Once一致性的計算能力，但在早期Flink版本中Checkpoint的性能在大規(guī)模數(shù)據(jù)量下存在一定瓶頸，阿里巴巴也在Checkpoint上進(jìn)行了大量改進(jìn)，比如：

增量Checkpoint機(jī)制：阿里巴巴生產(chǎn)環(huán)境中遇到大JOB有幾十TB
State是常事，做一次全量CP地動山搖，成本很高，因此阿里巴巴研發(fā)了增量Checkpoint機(jī)制，從此之后CP從暴風(fēng)驟雨變成了細(xì)水長流；

Checkpoint小文件合并：都是規(guī)模惹的禍，隨著整個集群Flink JOB越來越多，CP文件數(shù)也水漲船高，最后壓的HDFS NameNode不堪重負(fù)，阿里巴巴通過把若干CP小文件合并成一個大文件的組織方式，最終把NameNode的壓力減少了幾十倍。

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

Flink在設(shè)計上是一套批流統(tǒng)一的計算引擎，在使用過快如閃電的流計算之后，批用戶也開始有興趣入住Flink小區(qū)。但批計算也帶來了新的挑戰(zhàn)，首先在任務(wù)調(diào)度方面，阿里巴巴引入了更加靈活的調(diào)度機(jī)制，能夠根據(jù)任務(wù)之間的依賴關(guān)系進(jìn)行更加高效的調(diào)度；其次就是數(shù)據(jù)Shuffle，F(xiàn)link原生的Shuffle Service和TM綁定，任務(wù)執(zhí)行完之后要依舊保持TM無法釋放資源；還有就是原有的Batch shuffle沒有對文件進(jìn)行合并，所以基本無法在生產(chǎn)中使用。阿里巴巴開發(fā)了Yarn Shuffle Service功能的同時解決了以上兩個問題。在開發(fā)Yarn Shuffle Service的時候，阿里巴巴發(fā)現(xiàn)開發(fā)一套新的Shuffle Service非常不便，需要侵入Flink代碼的很多地方，為了讓其他開發(fā)者方便的擴(kuò)展不同Shuffle，阿里巴巴同時改造了Flink Shuffle架構(gòu)，讓Flink的Shuffle變成可插拔的架構(gòu)。目前阿里巴巴的搜索業(yè)務(wù)已經(jīng)在使用Flink Batch Job，并且已經(jīng)開始服務(wù)于生產(chǎn)。

經(jīng)過3年多打磨，Blink已經(jīng)在阿里巴巴開始茁壯生長，但是對Runtime的優(yōu)化和改進(jìn)是永無止境的，一大波改進(jìn)和優(yōu)化正在路上。

目前Flink已經(jīng)是一個主流的流計算引擎，社區(qū)下一步很重要的工作是讓Flink在批計算上有所突破，在更多的場景下落地，成為一種主流的批計算引擎。然后進(jìn)一步在流和批之間進(jìn)行無縫的切換，流和批的界限越來越模糊。用Flink,在一個計算中，既可以有流計算，又可以有批計算。

接下來阿里巴巴將致力于推動Flink在生態(tài)上得到更多語言的支持，不僅僅是Java、Scala語言，甚至是機(jī)器學(xué)習(xí)下用的Python、Go語言。

另一點(diǎn)不得不說AI，因?yàn)楝F(xiàn)在很多大數(shù)據(jù)計算的需求和數(shù)據(jù)量都是在支持很火爆的AI場景，所以Flink在流批生態(tài)完善的基礎(chǔ)上，將繼續(xù)完善上層的Machine Learning算法庫，同時Flink也會向更成熟的機(jī)器學(xué)習(xí)、深度學(xué)習(xí)去集成。比如可以做Tensorflow On Flink, 讓大數(shù)據(jù)的ETL數(shù)據(jù)處理和機(jī)器學(xué)習(xí)的Feature計算和特征計算，訓(xùn)練的計算等進(jìn)行集成，讓開發(fā)者能夠同時享受到多種生態(tài)給大家?guī)淼暮锰帯?/p>

最后，從生態(tài)、社區(qū)的活躍來說，阿里巴巴目前在推進(jìn)的一件事情是籌備2018年12月20日-21日在國家會議中心舉辦的首屆Flink Forward China峰會（千人規(guī)模），參與者將有機(jī)會了解阿里巴巴、騰訊、華為、滴滴、美團(tuán)、字節(jié)跳動等公司為何將Flink作為首選的流處理引擎。