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ù)進行批處理操作，將靜態(tài)數(shù)據(jù)抽象成分布式的數(shù)據(jù)集，用戶可以方便的采用Flink提供的各種操作符對分布式數(shù)據(jù)集進行各種操作，支持Java，Scala和Python。

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

3. Table API，對結(jié)構(gòu)化數(shù)據(jù)進行查詢操作，將結(jié)構(gòu)化數(shù)據(jù)抽象成關(guān)系表，并通過Flink提供的類SQL的DSL對關(guā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ù)一般被認為是兩種不同的任務(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ù)流切分成細粒度的batch數(shù)據(jù)，對于每一個batch數(shù)據(jù)，以此為輸入提交一個批處理Spark任務(wù)，所以Spark Streaming本質(zhì)上還是基于Spark批處理系統(tǒng)對流式數(shù)據(jù)進行處理，和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以固定的緩存塊為單位進行網(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)控，維護等成本。36大數(shù)據(jù)(http://www.36dsj.com/)

Flink流處理的容錯機制

對于一個分布式系統(tǒng)來說，單個進程或是節(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進行快照的時間間隔，當(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)進行快照，并存儲起來。當(dāng)一個操作符有多個輸入的時候，F(xiàn)link會將先抵達的快照標(biāo)記消息及其之后的消息緩存起來，當(dāng)所有的輸入中對應(yīng)該次快照的快照標(biāo)記消息全部抵達后，操作符對自己的狀態(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)的要求更嚴格，實現(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)的存儲可以是異步的，所以進行快照的過程基本上不會阻塞消息的處理，對消息的延遲不會產(chǎn)生負面的影響。

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ù)的時間越少，與吞吐量負相關(guān)。

Flink流處理的時間窗口

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

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

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

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

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

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

2. Event Time。消息自帶時間戳，根據(jù)消息的時間戳進行處理，確保時間戳在同一個時間窗口的所有消息一定會被正確處理。由于消息可能是亂序流入Task的，所以Task需要緩存當(dāng)前時間窗口消息處理的狀態(tài)，直到確認屬于該時間窗口的所有消息都被處理后，才可以釋放其狀態(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ù)。但是由于消息可能是亂序的，所以操作符無法直接確認何時所有屬于該時間窗口的消息全部流入此操作符。36大數(shù)據(jù)(http://www.36dsj.com/)

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

為了保證能夠處理所有屬于某個時間窗口的消息，操作符必須等到大于這個時間窗口的WaterMark之后，才能開始對該時間窗口的消息進行處理，相對于基于Operator Time的時間窗口，F(xiàn)link需要占用更多的內(nèi)存，且會直接影響消息處理的延遲時間。對此，一個可能的優(yōu)化措施是，對于聚合類的操作符，可能可以提前對部分消息進行聚合操作，當(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)生負面影響。

基于時間戳的排序

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

Flink基于時間戳進行排序的實現(xiàn)思路如下：排序操作符緩存所有流入的消息，當(dāng)其接收到WaterMark時，對時間戳小于該WaterMark的消息進行排序，并發(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還有一些其他很有意思的特性沒有詳細介紹，比如DataSet API級別的執(zhí)行計劃優(yōu)化器，原生的迭代操作符等，感興趣的讀者可以通過Flink的官網(wǎng)了解更多Flink的詳細內(nèi)容。希望通過本文的介紹能夠讓讀者對Flink項目能有更多的了解，也讓更多的人使用甚至參與到Flink項目中去。36大數(shù)據(jù)(http://www.36dsj.com/)

原文>>>