一文讀懂Apache Flink技術
本文是先介紹 Flink，再說 Flink的過去和現在
一、Flink介紹
Flink是一款分布式的計算引擎，它可以用來做批處理，即處理靜態的數據集、歷史的數據集;也可以用來做流處理，即實時地處理一些實時數據流，實時地產生數據的結果;也可以用來做一些基于事件的應用，比如說滴滴通過Flink CEP實現實時監測用戶及司機的行為流來判斷用戶或司機的行為是否正當。
總而言之，Flink是一個Stateful Computations Over Streams，即數據流上的有狀態的計算。這里面有兩個關鍵字，一個是Streams，Flink認為有界數據集是無界數據流的一種特例，所以說有界數據集也是一種數據流，事件流也是一種數據流。Everything is streams，即Flink可以用來處理任何的數據，可以支持批處理、流處理、AI、MachineLearning等等。
另外一個關鍵詞是Stateful，即有狀態計算。有狀態計算是最近幾年來越來越被用戶需求的一個功能。舉例說明狀態的含義，比如說一個網站一天內訪問UV數，那么這個UV數便為狀態。Flink提供了內置的對狀態的一致性的處理，即如果任務發生了Failover，其狀態不會丟失、不會被多算少算，同時提供了非常高的性能。
那Flink的受歡迎離不開它身上還有很多的標簽，其中包括性能優秀(尤其在流計算領域)、高可擴展性、支持容錯，是一種純內存式的一個計算引擎，做了內存管理方面的大量優化，另外也支持eventime的處理、支持超大狀態的Job(在阿里巴巴中作業的state大小超過TB的是非常常見的)、支持exactly-once的處理。
1.1 Flink基石
Flink之所以能這么流行，離不開它最重要的四個基石：Checkpoint、State、Time、Window。
首先是Checkpoint機制，這是Flink最重要的一個特性。Flink基于Chandy-Lamport算法實現了一個分布式的一致性的快照，從而提供了一致性的語義。Chandy-Lamport算法實際上在1985年的時候已經被提出來，但并沒有被很廣泛的應用，而Flink則把這個算法發揚光大了。Spark最近在實現Continue streaming，Continue streaming的目的是為了降低它處理的延時，其也需要提供這種一致性的語義，最終采用Chandy-Lamport這個算法，說明Chandy-Lamport算法在業界得到了一定的肯定。
提供了一致性的語義之后，Flink為了讓用戶在編程時能夠更輕松、更容易地去管理狀態，還提供了一套非常簡單明了的State API，包括里面的有ValueState、ListState、MapState，近期添加了BroadcastState，使用State API能夠自動享受到這種一致性的語義。
除此之外，Flink還實現了Watermark的機制，能夠支持基于事件的時間的處理，或者說基于系統時間的處理，能夠容忍數據的延時、容忍數據的遲到、容忍亂序的數據。
另外流計算中一般在對流數據進行操作之前都會先進行開窗，即基于一個什么樣的窗口上做這個計算。Flink提供了開箱即用的各種窗口，比如滑動窗口、滾動窗口、會話窗口以及非常靈活的自定義的窗口。
1.2 Flink API
Flink分層API主要有三層，如下圖：
最底層是ProcessFunction，它能夠提供非常靈活的功能，它能夠訪問各種各樣的State，用來注冊一些timer，利用timer回調的機制能夠實現一些基于事件驅動的一些應用。
之上是DataStream API，最上層是SQL/Table API的一種High-level API。
1.3 Flink的用途
Flink能用來做什么?回顧一下Flink up前幾站的分享，有非常多的嘉賓分享了他們在自己公司里面基于Flink做的一些實踐，包括攜程、唯品會、餓了么、滴滴、頭條等等。他們的應用場景包括實時的機器學習，實時的統計分析，實時的異常監測等等。這些實踐案例的共同點就是都用來做實時性的任務。
1.4 Flink Title的變化
早期Flink是這樣介紹自己的:“我是一個開源的流批統一的計算引擎”，當時跟Spark有點類似。后來Spark改成了一長串的文字，里面有各種各樣的形容詞：“我是一個分布式的、高性能的、高可用的、高精確的流計算系統”。最近Spark又進行了修改：“我是一個數據流上的有狀態的計算”。
通過觀察這個變化，可以發現Flink社區重心的變遷，即社區現在主要精力是放在打造它的流計算引擎上。先在流計算領域扎根，領先其他對手幾年，然后借助社區的力量壯大社區，再借助社區的力量擴展它的生態。
阿里巴巴Flink是這樣介紹自己的：“Flink是一個大數據量處理的統一的引擎”。這個“統一的引擎”包括流處理、批處理、AI、MachineLearning、圖計算等等。
二、Flink過去與現在
2.1 Flink High-Level API的歷史變遷
在Flink 1.0.0時期，Table API和CEP這兩個框架被首次加入到倉庫里面，同時社區對于SQL的需求很大。SQL和Table API非常相近，都是一種處理結構化數據的一種High-Level語言，實現上可以共用很多內容。所以在1.1.0里面，社區基于Apache Calcite對整個非Table的Module做了重大的重構，使得Table API和SQL共用了大部分的代碼，同時進行了支持。
在Flink 1.2.0時期，在Table API和SQL上支持Tumbling Window、Sliding Window、Session Window這些窗口。
在Flink 1.3.0時期，首次引用了Dynamic Table這個概念，借助Dynamic Table，流和批之間是可以相互進行轉換的。流可以是一張表，表也可以是一張流，這是流批統一的基礎之一。Retraction機制是Dynamic Table最重要的一個功能，基于Retraction才能夠正確地實現多級Application、多級Join，才能夠保證語意與結果的一個正確性。同時該版本支持了CEP算子的可控性。
在Flink 1.5.0時期，支持了Join操作，包括window Join以及非window Join，還添加了SQL CLI支持。SQL CLI提供了一個類似shell命令的對話框，可以交互式執行查詢。
2.2 Flink API的歷史變遷
- 在Flink 1.0.0時期，加入了State API，即ValueState、ReducingState、ListState等等。State API主要方便了DataStream用戶，使其能夠更加容易地管理狀態。
- 在Flink 1.1.0時期，提供了對SessionWindow以及遲到數據處理的支持。
- 在Flink 1.2.0時期，提供了ProcessFunction，一個Low-level的API?；赑rocessFunction用戶可以比較靈活地實現基于事件的一些應用。
- 在Flink 1.3.0時期，提供了Side outputs功能。一般算子的輸出只有一種輸出的類型，但是有些時候可能需要輸出另外的類型，比如把一些異常數據、遲到數據以側邊流的形式進行輸出，并交給異常節點進行下一步處理，這就是Side outputs。
- 在Flink 1.5.0時期，加入了BroadcastState。BroadcastState用來存儲上游被廣播過來的數據，這個節點上的很多N個并發上存在的BroadcastState里面的數據都是一模一樣的，因為它是從上游廣播來的?；谶@種State可以比較好地去解決不等值Join這種場景。比如一個Query里面寫的“SLECECT * FROM L JOIN R WHERE L.a > R.b”，也就是說我們需要把左表和右表里面所有A大于B的數據都關聯輸出出來。
在以前的實現中，由于沒有Join等值條件，就無法按照等值條件來做KeyBy的Shuffle，只能夠將所有的數據全部匯集到一個節點上，一個單并發的節點上進行處理，而這個單并發的節點就會成為整個Job的瓶頸。
而有了BroadcastState以后就可以做一些優化：因為左表數據量比較大，右表數據量比較小，所以選擇把右表進行廣播，把左表按照它某一個進行均勻分布的key，做keyby shuffle，shuffle到下游的N個Join的節點，Join的節點里面會存兩份State，左邊state和右邊state，左邊state用來存左邊數據流的state，是一個keyedState，因為它是按它某一個key做keyby分發下來的。右邊State是一個BroadcastState，所有的Join節點里面的BroadcastState里面存的數據都是一模一樣的，因為均為從上游廣播而來。
所有keyedState進行并發處理，之后將keyedState集合進行合并便等于左邊數據流的全集處理結果。于是便實現了這個Join節點的可擴充，通過增加join節點的并發，可以比較好地提升Job處理能力。除了不等值Join場景，BroadcastState還可以比較有效地解決像CAP上的動態規則。
在Flink 1.6.0時期，提供了State TTL參數、DataStream Interval Join功能。State TTL實現了在申請某個State時候可以在指定一個TTL參數，指定該state過了多久之后需要被系統自動清除。在這個版本之前，如果用戶想要實現這種狀態清理操作需要使用ProcessFunction注冊一個Timer，然后利用Timer的回調手動把這個State清除。從該版本開始，Flink框架可以基于TTL原生地解決這件事情。DataStream Interval Join功能即含有區間間隔的Join，比如說左流Join右流前后幾分鐘之內的數據，這種叫做Interval Join。
2.3 Flink Checkpoint & Recovery的歷史變遷
Checkpoint機制在Flink很早期的時候就已經支持，是Flink一個很核心的功能，Flink社區也一直致力于努力把Checkpoint效率提升，以及換成FailOver之后它的Recallable效率的提升。
在Flink 1.0.0時期，提供了RocksDB的支持，這個版本之前所有的狀態都只能存在進程的內存里面，這個內存總有存不下的一天，如果存不下則會發生OOM。如果想要存更多數據、更大量State就要用到RocksDB。RocksDB是一款基于文件的嵌入式數據庫，它會把數據存到磁盤，但是同時它又提供高效讀寫能力。所以使用RocksDB不會發生OOM這種事情。在Flink1.1.0里面，提供了純異步化的RocksDB的snapshot。以前版本在做RocksDB的snapshot時它會同步阻塞主數據流的處理，很影響吞吐量，即每當checkpoint時主數據流就會卡住。純異步化處理之后不會卡住數據流，于是吞吐量也得到了提升。
在Flink 1.2.0時期，引入了Rescalable keys和operate state的概念，它支持了一個Key State的可擴充以及operator state的可擴充。
在Flink 1.3.0時期，引入了增量的checkpoint這個比較重要的功能。只有基于增量的checkpoint才能更好地支持含有超大State的Job。在阿里內部，這種上TB的State是非常常見。如果每一次都把全量上TB的State都刷到遠程的HDFS上那么這個效率是很低下的。而增量checkpoint只是把checkpoint間隔新增的那些狀態發到遠程做存儲，每一次checkpoint發的數據就少了很多，效率得到提高。在這個版本里面還引入了一個細粒度的recovery，細粒度的recovery在做恢復的時候，有時不需要對整個Job做恢復，可能只需要恢復這個Job中的某一個子圖，這樣便能夠提高恢復效率。
在Flink 1.5.0時期，引入了Task local 的State的recovery。因為基于checkpoint機制，會把State持久化地存儲到某一個遠程存儲，比如HDFS，當發生Failover的時候需要重新把這個數據從遠程HDFS再download下來，如果這個狀態特別大那么該download操作的過程就會很漫長，導致Failover恢復所花的時間會很長。Task local state recovery提供的機制是當Job發生Failover之后，能夠保證該Job狀態在本地不會丟失，進行恢復時只需在本地直接恢復，不需從遠程HDFS重新把狀態download下來，于是就提升了Failover recovery的效率。
2.4 Flink Runtime的歷史變遷
Runtime的變遷歷史是非常重要的。
在Flink 1.2.0時期，提供了Async I/O功能。如果任務內部需要頻繁地跟外部存儲做查詢訪問，比如說查詢一個HBase表，在該版本之前每次查詢的操作都是阻塞的，會頻繁地被I/O的請求卡住。當加入異步I/O之后就可以同時地發起N個異步查詢的請求，這樣便提升了整個job的吞吐量，同時Async I/O又能夠保證該job的Async語義。
在Flink 1.3.0時期，引入了HistoryServer的模塊。HistoryServer主要功能是當job結束以后，它會把job的狀態以及信息都進行歸檔，方便后續開發人員做一些深入排查。
在Flink 1.4.0時期，提供了端到端的exactly once的語義保證，Flink中所謂exactly once一般是指Flink引擎本身的exactly once。如果要做到從輸入到處理再到輸出，整個端到端整體的exactly once的話，它需要輸出組件具備commit功能。在kafka老版本中不存在commit功能，從最近的1.1開始有了這個功能，于是Flink很快便實現了端到端exactly once。
在Flink 1.5.0時期，Flink首次對外正式地提到新的部署模型和處理模型。新的模型開發工作已經持續了很久，在阿里巴巴內部這個新的處理模型也已經運行了有兩年以上，該模型的實現對Flink內部代碼改動量特別大，可以說是自Flink項目建立以來，Runtime改動最大的一個改進。簡而言之，它的一個特性就是它可以使得在使用YARN、Mesos這種調度系統時，可以更加更好地動態分配資源、動態釋放資源、提高資源利用性，還有提供更好的jobs之間的隔離。最后是在這個版本中，Flink對其網絡站進行了一個基本重構。
2.5 Flink 網絡棧重構
在流計算中有兩個用來衡量性能的指標：延遲和吞吐。
一般來講如果想要更高吞吐就要犧牲一些延遲，如果想要更低的延遲就要犧牲一定的吞吐。但是網絡棧的重構卻實現了延遲和吞吐的同時提升，這主要得益于它兩方面的工作：第一個是基于信用的流控，另一個是基于事件的I/O。一個用來提高它的吞吐，另一個用來降低它的延遲。
在介紹流控之前需要先介紹一下現有的網絡棧。Flink中TaskManager就是用來管理各個task的角色，它是以進程為單位;task用來執行用戶代碼，以線程為單位。當tasks之間有數據傳輸的交互的時候就要建立網絡的連接，如果2秒之間都建立一個TCP連接的話，那么這個TCP連接會被嚴重浪費，所以Flink在兩個TaskManager之間建立一個TCP連接，即兩個進程之間只存在一個連接。各個task之間以TCP channel的方式來共享TCP的連接，這樣整個job中就不會有太多的TCP連接。
2.6 Flink 反壓
反壓的意思是當某一個task的處理性能跟不上輸入速率的時候，其輸入端的Buffer就會被填滿，當輸入端Buffer被填滿的時候就會導致TCP的讀取被暫停。TCP的讀取被暫停之后，就會導致上游輸出端的Buffer池越積越多，因為下游此時已經不再進行消費。
當上游輸出端的Buffer池也堆滿的時候， TCP通道就會被關閉，其內部所有的TCP channel也會被關閉。從而上游task就會逐級的向上游進行反壓，這是整體的反壓流程，所以說Flink以前的反壓機制是比較原生態、比較粗暴的，因為其控制力度很大，整個TCP中一旦某一個Task性能跟不上，就會把整個TCP連接關掉。如下圖所示：
右下角的task雖然處理跟不上了，但上面的task仍然可以繼續進行處理。左邊這些上游數據可以繼續發給右上角的task進行處理。但是由于現在整個的TCP連接都被關閉，導致右上角task同樣收不到數據，整體吞吐量實際上是下降的趨勢。為了優化這個功能就需要做到更加細密度的流控，目前是關閉整個TCP連接，優化措施就是需要對TCP channel進行控制，當某個task處理不過來時只需要該Task對應的TCP channel，其它TCP channel不受影響。優化實現方式就是基于信用的流控。
基于信用的流控的核心思想就是基于信用額度的消費。比如銀行做貸款，為了防止壞賬太多，它會對每一個人評估其信用額度，當發放貸款時貸款不會超過這個人能承受的額度?；谶@種方式，它能夠一方面不會產生太多壞賬，另一方面可以充分地把銀行的資金利用起來?；谛庞玫牧骺鼐褪腔谶@種思想，Flink中所謂的信用額度，就是指這個下游消費端的可用的Buffer數。如下圖：
該圖左邊是指發送端，有四個輸出的隊列，每個隊列里面的方塊代表輸出Buffer，即準備丟給下游處理的Buffer。右邊是消費端，消費端也有四個隊列，這四個隊列里面也有一些Buffer塊，這些Buffer塊是空閑的Buffer，準備用來接收上游發給自己的數據。
上面提到基于數據的流控中所謂的信用就是指這個消費端它可用的Buffer數，代表當前還能夠消費多少數據，消費端首先會向上游反饋當前的信用是多少， producer端只會向信用額度大于0的下游進行發送，對于信用額度如果為0的就不再發送數據。這樣整個網絡的利用率便得到了很大的提升，不會發生某些Buffer被長時間的停留在網絡的鏈路上的情況。
基于信用的流控主要有以下兩方面的優化提升：
- 一個是當某一個task發生反壓處理跟不上的時候，不會發生所有的task都卡住，這種做法使吞吐量得到了很大的提升，在阿里內部用雙11大屏作業進行測試，這種新的流控算法會得到20%的提升;
- 另一個是基于事件的I/O，Flink在網絡端寫數據時會先往一個Buffer塊里面寫數據，這個Buffer塊是一個32K的長度的單位，即32K的大小，當這個Buffer塊被填滿的時候就會輸出到網絡里面，或者如果數據流比較慢，沒辦法很快填滿的話，那么會等待一個超時，默認一個100毫秒，即如果100毫秒內還沒被填滿那么這個Buffer也會被輸出到網絡里面。此時若是在以前版本中Flink延遲可能是在100毫秒以內，最差的情況下是到100毫秒，因為需要到100毫秒等這個Buffer發出去。
如果要得到更低的延時，現在的做法就會將這個Buffer直接加入到輸出的隊列，但是還是保持繼續往這個Buffer塊里面寫數據，當網絡里面有容量時這個Buffer塊便會立刻被發出去，如果網絡現在也比較繁忙，那就繼續填充這個Buffer，這樣吞吐也會比較好一點?；谶@種算法，Flink的延時幾乎是完美的，可以看到它的曲線基本上是低于10毫秒的，這也充分利用了網絡的容量，幾乎對吞吐沒有影響。
