Flink項目是大數(shù)據(jù)處理領(lǐng)域最近冉冉升起的一顆新星，其不同于其他大數(shù)據(jù)項目的諸多特性吸引了越來越多的人關(guān)注Flink項目。本文將深入分析Flink一些關(guān)鍵的技術(shù)與特性，希望能夠幫助讀者對Flink有更加深入的了解，對其他大數(shù)據(jù)系統(tǒng)的開發(fā)者也能有所裨益。

注：本文假設(shè)讀者對MapReduce，Spark及Storm等大數(shù)據(jù)處理系統(tǒng)有基本了解，同時熟悉流處理與批處理的基本概念。36大數(shù)據(jù)(http://www.36dsj.com/)

Flink簡介

Flink的核心是一個流式的數(shù)據(jù)流執(zhí)行引擎，其針對數(shù)據(jù)流的分布式計算提供了數(shù)據(jù)分布，數(shù)據(jù)通信以及容錯機制等功能?；诹鲌?zhí)行引擎，F(xiàn)link提供了諸多更高抽象層的API以方便用戶編寫分布式任務(wù)：

1. DataSet API, 對靜態(tài)數(shù)據(jù)進(jìn)行批處理操作，將靜態(tài)數(shù)據(jù)抽象成分布式的數(shù)據(jù)集，用戶可以方便的采用Flink提供的各種操作符對分布式數(shù)據(jù)集進(jìn)行各種操作，支持Java，Scala和Python。

2. DataStream API，對數(shù)據(jù)流進(jìn)行流處理操作，將流式的數(shù)據(jù)抽象成分布式的數(shù)據(jù)流，用戶可以方便的采用Flink提供的各種操作符對分布式數(shù)據(jù)流進(jìn)行各種操作，支持Java和Scala。

3. Table API，對結(jié)構(gòu)化數(shù)據(jù)進(jìn)行查詢操作，將結(jié)構(gòu)化數(shù)據(jù)抽象成關(guān)系表，并通過Flink提供的類SQL的DSL對關(guān)系表進(jìn)行各種查詢操作，支持Java和Scala。

此外，F(xiàn)link還針對特定的應(yīng)用領(lǐng)域提供了領(lǐng)域庫，例如：

1. Flink ML，F(xiàn)link的機器學(xué)習(xí)庫，提供了機器學(xué)習(xí)Pipelines API以及很多的機器學(xué)習(xí)算法實現(xiàn)。

2. Gelly，F(xiàn)link的圖計算庫，提供了圖計算的相關(guān)API以及很多的圖計算算法實現(xiàn)。

Flink的技術(shù)棧如下圖所示：36大數(shù)據(jù)(http://www.36dsj.com/)

圖1 Flink技術(shù)棧

此外，F(xiàn)link也可以方便地和其他的Hadoop生態(tài)圈的項目集成，例如，F(xiàn)link可以讀取存儲在HDFS或HBase中的靜態(tài)數(shù)據(jù)，以Kafka作為流式的數(shù)據(jù)源，直接重用MapReduce/Storm代碼，或是通過YARN申請集群資源等等。

統(tǒng)一的批處理與流處理系統(tǒng)

在大數(shù)據(jù)處理領(lǐng)域，批處理任務(wù)與流處理任務(wù)一般被認(rèn)為是兩種不同的任務(wù)，一個大數(shù)據(jù)項目一般會被設(shè)計為只能處理其中一種任務(wù)，例如Apache Storm，Apache Smaza只支持流處理任務(wù)，而Aapche MapReduce， Apache Tez，Apache Spark只支持批處理任務(wù)。

Spark Streaming是Apache Spark之上支持流處理任務(wù)的子系統(tǒng)，看似一個特例，實則不然。Spark Streaming采用了一種micro-batch的架構(gòu)，即將輸入的數(shù)據(jù)流切分成細(xì)粒度的batch數(shù)據(jù)，對于每一個batch數(shù)據(jù)，以此為輸入提交一個批處理Spark任務(wù)，所以Spark Streaming本質(zhì)上還是基于Spark批處理系統(tǒng)對流式數(shù)據(jù)進(jìn)行處理，和Apache Storm，Apache Smaza等完全流式的數(shù)據(jù)處理方式完全不同。Flink能夠同時處理批處理任務(wù)與流處理任務(wù)，其靈活的執(zhí)行引擎支持完全原生的批量的數(shù)據(jù)處理和流式的數(shù)據(jù)處理。

在執(zhí)行引擎這一層，流處理系統(tǒng)與批處理系統(tǒng)***的不同在于節(jié)點間數(shù)據(jù)傳輸?shù)姆绞?。對于一個流處理系統(tǒng)，其節(jié)點間數(shù)據(jù)傳輸?shù)臉?biāo)準(zhǔn)模型是：當(dāng)一條數(shù)據(jù)被處理完成后，序列化到緩存中，然后立刻通過網(wǎng)絡(luò)傳輸?shù)较乱粋€節(jié)點，由下一個節(jié)點繼續(xù)處理。而對于一個批處理系統(tǒng)，其節(jié)點間數(shù)據(jù)傳輸?shù)臉?biāo)準(zhǔn)模型是：當(dāng)一條數(shù)據(jù)被處理完成后，序列化到緩存中，并不會立刻通過網(wǎng)絡(luò)傳輸?shù)较乱粋€節(jié)點，當(dāng)緩存寫滿，就持久化到本地硬盤上，當(dāng)所有數(shù)據(jù)都被處理完成后，才開始將處理后的數(shù)據(jù)通過網(wǎng)絡(luò)傳輸?shù)较乱粋€節(jié)點。36大數(shù)據(jù)(http://www.36dsj.com/)

這兩種數(shù)據(jù)傳輸模式是兩個極端，對應(yīng)的是流處理系統(tǒng)對低延遲的要求和批處理系統(tǒng)對高吞吐量的要求。Flink的執(zhí)行引擎采用了一種十分靈活的方式，同時支持了這兩種數(shù)據(jù)傳輸模型。Flink以固定的緩存塊為單位進(jìn)行網(wǎng)絡(luò)數(shù)據(jù)傳輸，用戶可以通過緩存塊超時值指定緩存塊的傳輸時機。如果緩存塊的超時值為0，則Flink的數(shù)據(jù)傳輸方式類似上面提到的流處理系統(tǒng)的標(biāo)準(zhǔn)模型，此時系統(tǒng)可以獲得***的處理延遲。

如果緩存塊的超時值為***大，則Flink的數(shù)據(jù)傳輸方式類似上面提到的批處理系統(tǒng)的標(biāo)準(zhǔn)模型，此時系統(tǒng)可以獲得***的處理吞吐量。同時緩存塊的超時值也可以設(shè)置為0到***大之間的任意值。緩存塊的超時閾值越小，則Flink流處理執(zhí)行引擎的數(shù)據(jù)處理延遲越低，但吞吐量也會越低，緩存塊的超時閾值越大時，則反之。通過調(diào)整緩存塊的超時閾值，用戶可根據(jù)自己的需要靈活的權(quán)衡Flink的延遲和吞吐量。

圖2 Flink執(zhí)行引擎數(shù)據(jù)傳輸模式

在統(tǒng)一的流式執(zhí)行引擎的基礎(chǔ)上，F(xiàn)link同時支持了流處理系統(tǒng)與批處理系統(tǒng)，并且保證了其流處理系統(tǒng)與批處理系統(tǒng)的性能(延遲，吞吐量等)，相對于其他原生的流處理與批處理系統(tǒng)，并沒有因為統(tǒng)一的執(zhí)行引擎而受到影響。用戶可以在Flink上同時執(zhí)行批處理任務(wù)與流處理任務(wù)，這大大減輕了用戶安裝，部署，監(jiān)控，維護(hù)等成本。36大數(shù)據(jù)(http://www.36dsj.com/)

Flink流處理的容錯機制

對于一個分布式系統(tǒng)來說，單個進(jìn)程或是節(jié)點崩潰導(dǎo)致整個Job失敗是經(jīng)常發(fā)生的事情，在異常發(fā)生的時候不會丟失用戶數(shù)據(jù)，并能夠自動恢復(fù)是分布式系統(tǒng)的需要支持的特性之一。本節(jié)主要介紹Flink流處理系統(tǒng)對于任務(wù)級別的容錯機制。

批處理系統(tǒng)比較容易實現(xiàn)容錯機制，由于文件可以重復(fù)訪問，當(dāng)某個任務(wù)失敗后，重啟該任務(wù)即可。但是在流處理系統(tǒng)中，由于數(shù)據(jù)源是***的數(shù)據(jù)流，一個流處理任務(wù)甚至可能會執(zhí)行幾個月，將所有數(shù)據(jù)緩存或是持久化，留待以后重復(fù)訪問基本上是不可行的。Flink基于分布式快照與可部分重發(fā)的數(shù)據(jù)源實現(xiàn)了容錯，用戶可自定義對整個Job進(jìn)行快照的時間間隔，當(dāng)出現(xiàn)任務(wù)失敗時，F(xiàn)link將整個Job恢復(fù)到最近一次快照的狀態(tài)，并從數(shù)據(jù)源重發(fā)快照之后的數(shù)據(jù)。

Flink的分布式快照的實現(xiàn)借鑒了Chandy和Lamport在1985年發(fā)表的一篇關(guān)于分布式快照的論文，其實現(xiàn)的主要思想如下：

按照用戶自定義的分布式快照間隔時間，F(xiàn)link會在定時在所有數(shù)據(jù)源中插入一種特殊的快照標(biāo)記消息，這些快照標(biāo)記消息和其他消息一樣在DAG中流動，但是不會被用戶定義的業(yè)務(wù)邏輯所處理，每一個快照標(biāo)記消息都將其所在的數(shù)據(jù)流分成兩部分：本次快照數(shù)據(jù)和下次快照數(shù)據(jù)。36大數(shù)據(jù)(http://www.36dsj.com/)

圖3 Flink包含快照標(biāo)記消息的消息流

快照標(biāo)記消息沿著DAG流經(jīng)各個操作符，當(dāng)操作符處理到快照標(biāo)記消息時，會對自己的狀態(tài)進(jìn)行快照，并存儲起來。當(dāng)一個操作符有多個輸入的時候，F(xiàn)link會將先抵達(dá)的快照標(biāo)記消息及其之后的消息緩存起來，當(dāng)所有的輸入中對應(yīng)該次快照的快照標(biāo)記消息全部抵達(dá)后，操作符對自己的狀態(tài)快照并存儲，之后處理所有快照標(biāo)記消息之后的已緩存消息。操作符對自己的狀態(tài)快照并存儲可以是異步與增量的操作，并不需要阻塞消息的處理。分布式快照的流程如下圖所示：

圖4 Flink分布式快照流程圖

當(dāng)所有的Data Sink(終點操作符)都收到快照標(biāo)記信息并對自己的狀態(tài)快照和存儲后，整個分布式快照就完成了，同時通知數(shù)據(jù)源釋放該快照標(biāo)記消息之前的所有消息。若之后發(fā)生節(jié)點崩潰等異常情況時，只需要恢復(fù)之前存儲的分布式快照狀態(tài)，并從數(shù)據(jù)源重發(fā)該快照以后的消息就可以了。

Exactly-Once是流處理系統(tǒng)需要支持的一個非常重要的特性，它保證每一條消息被流處理系統(tǒng)處理一次，且僅被處理一次，許多流處理任務(wù)的業(yè)務(wù)邏輯都依賴于Exactly-Once特性。相對于At-Least-Once或是At-Most-Once, Exactly-Once特性對流處理系統(tǒng)的要求更嚴(yán)格，實現(xiàn)也更困難。Flink基于分布式快照實現(xiàn)了Exactly-Once特性。36大數(shù)據(jù)(http://www.36dsj.com/)

相對于其他流處理系統(tǒng)的容錯方案，F(xiàn)link基于分布式快照的方案在功能和性能方面都具有很多優(yōu)點，包括：

1. 低延遲。由于操作符狀態(tài)的存儲可以是異步的，所以進(jìn)行快照的過程基本上不會阻塞消息的處理，對消息的延遲不會產(chǎn)生負(fù)面的影響。

2. 高吞吐量。當(dāng)操作符狀態(tài)較少時，對吞吐量基本沒有影響。當(dāng)操作符狀態(tài)較多時，相對于其他的容錯機制，分布式快照的時間間隔是用戶自定義的，所以用戶可以權(quán)衡錯誤恢復(fù)時間和吞吐量的要求，調(diào)整分布式快照的時間間隔。

3. 與業(yè)務(wù)邏輯的隔離。Flink的分布式快照機制與用戶的業(yè)務(wù)邏輯是完全隔離的，用戶的業(yè)務(wù)邏輯不會依賴或是對分布式快照產(chǎn)生任何影響。

4. 錯誤恢復(fù)代價。分布式快照的時間間隔越短，錯誤恢復(fù)的時間越少，與吞吐量負(fù)相關(guān)。

Flink流處理的時間窗口

對于流處理系統(tǒng)來說，流入的消息是***的，所以對于聚合或是連接等操作，流處理系統(tǒng)需要對流入的消息進(jìn)行分段，然后基于每一段數(shù)據(jù)進(jìn)行聚合或是連接等操作。消息的分段即稱為窗口，流處理系統(tǒng)支持的窗口有很多類型，最常見的就是時間窗口，基于時間間隔對消息進(jìn)行分段處理。本節(jié)主要介紹Flink流處理系統(tǒng)支持的各種時間窗口。

對于目前大部分流處理系統(tǒng)來說，時間窗口一般是根據(jù)Task所在節(jié)點的本地時鐘來進(jìn)行切分，這種方式實現(xiàn)起來比較容易，不會阻塞消息處理。但是可能無法滿足某些應(yīng)用的要求，例如：

1. 消息本身帶有時間戳，用戶希望按照消息本身的時間特性進(jìn)行分段處理。

2. 由于不同節(jié)點的時鐘可能不同，以及消息在流經(jīng)各個節(jié)點時延遲不同，在某個節(jié)點屬于同一個時間窗口處理的消息，流到下一個節(jié)點時可能被切分到不同的時間窗口中，從而產(chǎn)生不符合預(yù)期的結(jié)果。

Flink支持三種類型的時間窗口，分別適用于用戶對于時間窗口不同類型的要求：

1. Operator Time。根據(jù)Task所在節(jié)點的本地時鐘來進(jìn)行切分的時間窗口。

2. Event Time。消息自帶時間戳，根據(jù)消息的時間戳進(jìn)行處理，確保時間戳在同一個時間窗口的所有消息一定會被正確處理。由于消息可能是亂序流入Task的，所以Task需要緩存當(dāng)前時間窗口消息處理的狀態(tài)，直到確認(rèn)屬于該時間窗口的所有消息都被處理后，才可以釋放其狀態(tài)。如果亂序的消息延遲很高的話，會影響分布式系統(tǒng)的吞吐量和延遲。

3. Ingress Time。有時消息本身并不帶有時間戳信息，但用戶依然希望按照消息而不是節(jié)點時鐘劃分時間窗口(例如，避免上面提到的第二個問題)。此時可以在消息源流入Flink流處理系統(tǒng)時，自動生成增量的時間戳賦予消息，之后處理的流程與Event Time相同。Ingress Time可以看成是Event Time的一個特例，由于其在消息源處時間戳一定是有序的，所以在流處理系統(tǒng)中，相對于Event Time，其亂序的消息延遲不會很高，因此對Flink分布式系統(tǒng)的吞吐量和延遲的影響也會更小。

Event Time時間窗口的實現(xiàn)

Flink借鑒了Google的MillWheel項目，通過WaterMark來支持基于Event Time時間窗口。

當(dāng)操作符通過基于Event Time的時間窗口來處理數(shù)據(jù)時，它必須在確定所有屬于該時間窗口的消息全部流入此操作符后，才能開始處理數(shù)據(jù)。但是由于消息可能是亂序的，所以操作符無法直接確認(rèn)何時所有屬于該時間窗口的消息全部流入此操作符。36大數(shù)據(jù)(http://www.36dsj.com/)

WaterMark包含一個時間戳，F(xiàn)link使用WaterMark標(biāo)記所有小于該時間戳的消息都已流入，F(xiàn)link的數(shù)據(jù)源在確認(rèn)所有小于某個時間戳的消息都已輸出到Flink流處理系統(tǒng)后，會生成一個包含該時間戳的WaterMark，插入到消息流中輸出到Flink流處理系統(tǒng)中，F(xiàn)link操作符按照時間窗口緩存所有流入的消息，當(dāng)操作符處理到WaterMark時，它對所有小于該WaterMark時間戳的時間窗口的數(shù)據(jù)進(jìn)行處理并發(fā)送到下一個操作符節(jié)點，然后也將WaterMark發(fā)送到下一個操作符節(jié)點。

為了保證能夠處理所有屬于某個時間窗口的消息，操作符必須等到大于這個時間窗口的WaterMark之后，才能開始對該時間窗口的消息進(jìn)行處理，相對于基于Operator Time的時間窗口，F(xiàn)link需要占用更多的內(nèi)存，且會直接影響消息處理的延遲時間。對此，一個可能的優(yōu)化措施是，對于聚合類的操作符，可能可以提前對部分消息進(jìn)行聚合操作，當(dāng)有屬于該時間窗口的新消息流入時，基于之前的部分聚合結(jié)果繼續(xù)計算，這樣的話，只需緩存中間計算結(jié)果即可，無需緩存該時間窗口的所有消息。

對于基于Event Time時間窗口的操作符來說，流入WaterMark的時間戳與當(dāng)前節(jié)點的時鐘一致是最簡單理想的狀況了，但是在實際環(huán)境中是不可能的，由于消息的亂序以及前面節(jié)點處理效率的不同，總是會有某些消息流入時間大于其本身的時間戳，真實WaterMark時間戳與理想情況下WaterMark時間戳的差別稱為Time Skew，如下圖所示：

圖5 WaterMark的Time Skew圖

Time Skew決定了該WaterMark與上一個WaterMark之間的時間窗口所有數(shù)據(jù)需要緩存的時間，Time Skew時間越長，該時間窗口數(shù)據(jù)的延遲越長，占用內(nèi)存的時間也越長，同時會對流處理系統(tǒng)的吞吐量產(chǎn)生負(fù)面影響。

基于時間戳的排序

在流處理系統(tǒng)中，由于流入的消息是***的，所以對消息進(jìn)行排序基本上被認(rèn)為是不可行的。但是在Flink流處理系統(tǒng)中，基于WaterMark，F(xiàn)link實現(xiàn)了基于時間戳的全局排序。

Flink基于時間戳進(jìn)行排序的實現(xiàn)思路如下：排序操作符緩存所有流入的消息，當(dāng)其接收到WaterMark時，對時間戳小于該WaterMark的消息進(jìn)行排序，并發(fā)送到下一個節(jié)點，在此排序操作符中釋放所有時間戳小于該WaterMark的消息，繼續(xù)緩存流入的消息，等待下一個WaterMark觸發(fā)下一次排序。

由于WaterMark保證了其之后不會出現(xiàn)時間戳比它小的消息，所以可以保證排序的正確性。需要注意的是，如果排序操作符有多個節(jié)點，只能保證每個節(jié)點的流出消息是有序的，節(jié)點之間的消息不能保證有序，要實現(xiàn)全局有序，則只能有一個排序操作符節(jié)點。

通過支持基于Event Time的消息處理，F(xiàn)link擴展了其流處理系統(tǒng)的應(yīng)用范圍，使得更多的流處理任務(wù)可以通過Flink來執(zhí)行。

定制的內(nèi)存管理

略，請參考上篇文章：脫離JVM? Hadoop生態(tài)圈的掙扎與演化

總結(jié)

本文主要介紹了Flink項目的一些關(guān)鍵特性，F(xiàn)link是一個擁有諸多特色的項目，包括其統(tǒng)一的批處理和流處理執(zhí)行引擎，通用大數(shù)據(jù)計算框架與傳統(tǒng)數(shù)據(jù)庫系統(tǒng)的技術(shù)結(jié)合，以及流處理系統(tǒng)的諸多技術(shù)創(chuàng)新等，因為篇幅有限，F(xiàn)link還有一些其他很有意思的特性沒有詳細(xì)介紹，比如DataSet API級別的執(zhí)行計劃優(yōu)化器，原生的迭代操作符等，感興趣的讀者可以通過Flink的官網(wǎng)了解更多Flink的詳細(xì)內(nèi)容。希望通過本文的介紹能夠讓讀者對Flink項目能有更多的了解，也讓更多的人使用甚至參與到Flink項目中去。36大數(shù)據(jù)(http://www.36dsj.com/)

原文>>>