Apache Flink(下簡稱Flink)項目是大數(shù)據(jù)處理領(lǐng)域最近冉冉升起的一顆新星，其不同于其他大數(shù)據(jù)項目的諸多特性吸引了越來越多人的關(guān)注。本文將深入分析Flink的一些關(guān)鍵技術(shù)與特性，希望能夠幫助讀者對Flink有更加深入的了解，對其他大數(shù)據(jù)系統(tǒng)開發(fā)者也能有所裨益。本文假設(shè)讀者已對MapReduce、Spark及Storm等大數(shù)據(jù)處理框架有所了解，同時熟悉流處理與批處理的基本概念。

Flink簡介

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

• DataSet API， 對靜態(tài)數(shù)據(jù)進行批處理操作，將靜態(tài)數(shù)據(jù)抽象成分布式的數(shù)據(jù)集，用戶可以方便地使用Flink提供的各種操作符對分布式數(shù)據(jù)集進行處理，支持Java、Scala和Python。
• DataStream API，對數(shù)據(jù)流進行流處理操作，將流式的數(shù)據(jù)抽象成分布式的數(shù)據(jù)流，用戶可以方便地對分布式數(shù)據(jù)流進行各種操作，支持Java和Scala。
• Table API，對結(jié)構(gòu)化數(shù)據(jù)進行查詢操作，將結(jié)構(gòu)化數(shù)據(jù)抽象成關(guān)系表，并通過類SQL的DSL對關(guān)系表進行各種查詢操作，支持Java和Scala。

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

• Flink ML，F(xiàn)link的機器學習庫，提供了機器學習Pipelines API并實現(xiàn)了多種機器學習算法。
• Gelly，F(xiàn)link的圖計算庫，提供了圖計算的相關(guān)API及多種圖計算算法實現(xiàn)。

Flink的技術(shù)棧如圖1所示：

此外，F(xiàn)link也可以方便地和Hadoop生態(tài)圈中其他項目集成，例如Flink可以讀取存儲在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，并為每一個batch數(shù)據(jù)提交一個批處理的Spark任務(wù)，所以Spark Streaming本質(zhì)上還是基于Spark批處理系統(tǒng)對流式數(shù)據(jù)進行處理，和Apache Storm、Apache Smaza等完全流式的數(shù)據(jù)處理方式完全不同。通過其靈活的執(zhí)行引擎，F(xiàn)link能夠同時支持批處理任務(wù)與流處理任務(wù)。

在執(zhí)行引擎這一層，流處理系統(tǒng)與批處理系統(tǒng)最大不同在于節(jié)點間的數(shù)據(jù)傳輸方式。對于一個流處理系統(tǒng)，其節(jié)點間數(shù)據(jù)傳輸?shù)臉藴誓Ｐ褪牵寒斠粭l數(shù)據(jù)被處理完成后，序列化到緩存中，然后立刻通過網(wǎng)絡(luò)傳輸?shù)较乱粋€節(jié)點，由下一個節(jié)點繼續(xù)處理。而對于一個批處理系統(tǒng)，其節(jié)點間數(shù)據(jù)傳輸?shù)臉藴誓Ｐ褪牵寒斠粭l數(shù)據(jù)被處理完成后，序列化到緩存中，并不會立刻通過網(wǎng)絡(luò)傳輸?shù)较乱粋€節(jié)點，當緩存寫滿，就持久化到本地硬盤上，當所有數(shù)據(jù)都被處理完成后，才開始將處理后的數(shù)據(jù)通過網(wǎng)絡(luò)傳輸?shù)较乱粋€節(jié)點。這兩種數(shù)據(jù)傳輸模式是兩個極端，對應的是流處理系統(tǒng)對低延遲的要求和批處理系統(tǒng)對高吞吐量的要求。Flink的執(zhí)行引擎采用了一種十分靈活的方式，同時支持了這兩種數(shù)據(jù)傳輸模型。Flink以固定的緩存塊為單位進行網(wǎng)絡(luò)數(shù)據(jù)傳輸，用戶可以通過緩存塊超時值指定緩存塊的傳輸時機。如果緩存塊的超時值為0，則Flink的數(shù)據(jù)傳輸方式類似上文所提到流處理系統(tǒng)的標準模型，此時系統(tǒng)可以獲得最低的處理延遲。如果緩存塊的超時值為無限大，則Flink的數(shù)據(jù)傳輸方式類似上文所提到批處理系統(tǒng)的標準模型，此時系統(tǒng)可以獲得最高的吞吐量。同時緩存塊的超時值也可以設(shè)置為0到無限大之間的任意值。緩存塊的超時閾值越小，則Flink流處理執(zhí)行引擎的數(shù)據(jù)處理延遲越低，但吞吐量也會降低，反之亦然。通過調(diào)整緩存塊的超時閾值，用戶可根據(jù)需求靈活地權(quán)衡系統(tǒng)延遲和吞吐量。

在統(tǒng)一的流式執(zhí)行引擎基礎(chǔ)上，F(xiàn)link同時支持了流計算和批處理，并對性能(延遲、吞吐量等)有所保障。相對于其他原生的流處理與批處理系統(tǒng)，并沒有因為統(tǒng)一執(zhí)行引擎而受到影響從而大幅度減輕了用戶安裝、部署、監(jiān)控、維護等成本。

Flink流處理的容錯機制

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

批處理系統(tǒng)比較容易實現(xiàn)容錯機制，由于文件可以重復訪問，當某個任務(wù)失敗后，重啟該任務(wù)即可。但是到了流處理系統(tǒng)，由于數(shù)據(jù)源是無限的數(shù)據(jù)流，從而導致一個流處理任務(wù)執(zhí)行幾個月的情況，將所有數(shù)據(jù)緩存或是持久化，留待以后重復訪問基本上是不可行的。Flink基于分布式快照與可部分重發(fā)的數(shù)據(jù)源實現(xiàn)了容錯。用戶可自定義對整個Job進行快照的時間間隔，當任務(wù)失敗時，F(xiàn)link會將整個Job恢復到最近一次快照，并從數(shù)據(jù)源重發(fā)快照之后的數(shù)據(jù)。Flink的分布式快照實現(xiàn)借鑒了Chandy和Lamport在1985年發(fā)表的一篇關(guān)于分布式快照的論文，其實現(xiàn)的主要思想如下：

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

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

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

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

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

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

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

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

Flink流處理的時間窗口

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

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

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

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

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

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

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

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的時間窗口。

當操作符通過基于Event Time的時間窗口來處理數(shù)據(jù)時，它必須在確定所有屬于該時間窗口的消息全部流入此操作符后才能開始數(shù)據(jù)處理。但是由于消息可能是亂序的，所以操作符無法直接確認何時所有屬于該時間窗口的消息全部流入此操作符。WaterMark包含一個時間戳，F(xiàn)link使用WaterMark標記所有小于該時間戳的消息都已流入，F(xiàn)link的數(shù)據(jù)源在確認所有小于某個時間戳的消息都已輸出到Flink流處理系統(tǒng)后，會生成一個包含該時間戳的WaterMark，插入到消息流中輸出到Flink流處理系統(tǒng)中，F(xiàn)link操作符按照時間窗口緩存所有流入的消息，當操作符處理到WaterMark時，它對所有小于該WaterMark時間戳的時間窗口數(shù)據(jù)進行處理并發(fā)送到下一個操作符節(jié)點，然后也將WaterMark發(fā)送到下一個操作符節(jié)點。

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

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

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)了基于時間戳的全局排序。排序的實現(xiàn)思路如下：排序操作符緩存所有流入的消息，當其接收到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)的應用范圍，使得更多的流處理任務(wù)可以通過Flink來執(zhí)行。

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

Flink項目基于Java及Scala等JVM語言，JVM本身作為一個各種類型應用的執(zhí)行平臺，其對Java對象的管理也是基于通用的處理策略，其垃圾回收器通過估算Java對象的生命周期對Java對象進行有效率的管理。

針對不同類型的應用，用戶可能需要針對該類型應用的特點，配置針對性的JVM參數(shù)更有效率的管理Java對象，從而提高性能。這種JVM調(diào)優(yōu)的黑魔法需要用戶對應用本身及JVM的各參數(shù)有深入了解，極大地提高了分布式計算平臺的調(diào)優(yōu)門檻。Flink框架本身了解計算邏輯每個步驟的數(shù)據(jù)傳輸，相比于JVM垃圾回收器，其了解更多的Java對象生命周期，從而為更有效率地管理Java對象提供了可能。

JVM存在的問題

Java對象開銷

相對于c/c++等更加接近底層的語言，Java對象的存儲密度相對偏低，例如[1]，“abcd”這樣簡單的字符串在UTF-8編碼中需要4個字節(jié)存儲，但采用了UTF-16編碼存儲字符串的Java則需要8個字節(jié)，同時Java對象還有header等其他額外信息，一個4字節(jié)字符串對象在Java中需要48字節(jié)的空間來存儲。對于大部分的大數(shù)據(jù)應用，內(nèi)存都是稀缺資源，更有效率地內(nèi)存存儲，意味著CPU數(shù)據(jù)訪問吞吐量更高，以及更少磁盤落地的存在。

對象存儲結(jié)構(gòu)引發(fā)的cache miss

為了緩解CPU處理速度與內(nèi)存訪問速度的差距[2]，現(xiàn)代CPU數(shù)據(jù)訪問一般都會有多級緩存。當從內(nèi)存加載數(shù)據(jù)到緩存時，一般是以cache line為單位加載數(shù)據(jù)，所以當CPU訪問的數(shù)據(jù)如果是在內(nèi)存中連續(xù)存儲的話，訪問的效率會非常高。如果CPU要訪問的數(shù)據(jù)不在當前緩存所有的cache line中，則需要從內(nèi)存中加載對應的數(shù)據(jù)，這被稱為一次cache miss。當cache miss非常高的時候，CPU大部分的時間都在等待數(shù)據(jù)加載，而不是真正的處理數(shù)據(jù)。Java對象并不是連續(xù)的存儲在內(nèi)存上，同時很多的Java數(shù)據(jù)結(jié)構(gòu)的數(shù)據(jù)聚集性也不好。

大數(shù)據(jù)的垃圾回收

Java的垃圾回收機制一直讓Java開發(fā)者又愛又恨，一方面它免去了開發(fā)者自己回收資源的步驟，提高了開發(fā)效率，減少了內(nèi)存泄漏的可能，另一方面垃圾回收也是Java應用的不定時炸彈，有時秒級甚至是分鐘級的垃圾回收極大影響了Java應用的性能和可用性。在時下數(shù)據(jù)中心，大容量內(nèi)存得到了廣泛的應用，甚至出現(xiàn)了單臺機器配置TB內(nèi)存的情況，同時，大數(shù)據(jù)分析通常會遍歷整個源數(shù)據(jù)集，對數(shù)據(jù)進行轉(zhuǎn)換、清洗、處理等步驟。在這個過程中，會產(chǎn)生海量的Java對象，JVM的垃圾回收執(zhí)行效率對性能有很大影響。通過JVM參數(shù)調(diào)優(yōu)提高垃圾回收效率需要用戶對應用和分布式計算框架以及JVM的各參數(shù)有深入了解，而且有時候這也遠遠不夠。

OOM問題

OutOfMemoryError是分布式計算框架經(jīng)常會遇到的問題，當JVM中所有對象大小超過分配給JVM的內(nèi)存大小時，就會出現(xiàn)OutOfMemoryError錯誤，JVM崩潰，分布式框架的健壯性和性能都會受到影響。通過JVM管理內(nèi)存，同時試圖解決OOM問題的應用，通常都需要檢查Java對象的大小，并在某些存儲Java對象特別多的數(shù)據(jù)結(jié)構(gòu)中設(shè)置閾值進行控制。但是JVM并沒有提供官方檢查Java對象大小的工具，第三方的工具類庫可能無法準確通用地確定Java對象大小[6]。侵入式的閾值檢查也會為分布式計算框架的實現(xiàn)增加很多額外與業(yè)務(wù)邏輯無關(guān)的代碼。

Flink的處理策略

為了解決以上提到的問題，高性能分布式計算框架通常需要以下技術(shù)：

定制的序列化工具。顯式內(nèi)存管理的前提步驟就是序列化，將Java對象序列化成二進制數(shù)據(jù)存儲在內(nèi)存上(on heap或是off-heap)。通用的序列化框架，如Java默認使用java.io.Serializable將Java對象及其成員變量的所有元信息作為其序列化數(shù)據(jù)的一部分，序列化后的數(shù)據(jù)包含了所有反序列化所需的信息。這在某些場景中十分必要，但是對于Flink這樣的分布式計算框架來說，這些元數(shù)據(jù)信息可能是冗余數(shù)據(jù)。定制的序列化框架，如Hadoop的org.apache.hadoop.io.Writable需要用戶實現(xiàn)該接口，并自定義類的序列化和反序列化方法。這種方式效率最高，但需要用戶額外的工作，不夠友好。

顯式的內(nèi)存管理。一般通用的做法是批量申請和釋放內(nèi)存，每個JVM實例有一個統(tǒng)一的內(nèi)存管理器，所有內(nèi)存的申請和釋放都通過該內(nèi)存管理器進行。這可以避免常見的內(nèi)存碎片問題，同時由于數(shù)據(jù)以二進制的方式存儲，可以大大減輕垃圾回收壓力。

緩存友好的數(shù)據(jù)結(jié)構(gòu)和算法。對于計算密集的數(shù)據(jù)結(jié)構(gòu)和算法，直接操作序列化后的二進制數(shù)據(jù)，而不是將對象反序列化后再進行操作。同時，只將操作相關(guān)的數(shù)據(jù)連續(xù)存儲，可以最大化的利用L1/L2/L3緩存，減少Cache miss的概率，提升CPU計算的吞吐量。以排序為例，由于排序的主要操作是對Key進行對比，如果將所有排序數(shù)據(jù)的Key與Value分開并對Key連續(xù)存儲，那么訪問Key時的Cache命中率會大大提高。

定制的序列化工具

分布式計算框架可以使用定制序列化工具的前提是要待處理數(shù)據(jù)流通常是同一類型，由于數(shù)據(jù)集對象的類型固定，從而可以只保存一份對象Schema信息，節(jié)省大量的存儲空間。同時，對于固定大小的類型，也可通過固定的偏移位置存取。在需要訪問某個對象成員變量時，通過定制的序列化工具，并不需要反序列化整個Java對象，而是直接通過偏移量，從而只需要反序列化特定的對象成員變量。如果對象的成員變量較多時，能夠大大減少Java對象的創(chuàng)建開銷，以及內(nèi)存數(shù)據(jù)的拷貝大小。Flink數(shù)據(jù)集都支持任意Java或是Scala類型，通過自動生成定制序列化工具，既保證了API接口對用戶友好(不用像Hadoop那樣數(shù)據(jù)類型需要繼承實現(xiàn)org.apache.hadoop.io.Writable接口)，也達到了和Hadoop類似的序列化效率。

Flink對數(shù)據(jù)集的類型信息進行分析，然后自動生成定制的序列化工具類。Flink支持任意的Java或是Scala類型，通過Java Reflection框架分析基于Java的Flink程序UDF(User Define Function)的返回類型的類型信息，通過Scala Compiler分析基于Scala的Flink程序UDF的返回類型的類型信息。類型信息由TypeInformation類表示，這個類有諸多具體實現(xiàn)類，例如：

• BasicTypeInfo任意Java基本類型(裝包或未裝包)和String類型。
• BasicArrayTypeInfo任意Java基本類型數(shù)組(裝包或未裝包)和String數(shù)組。
• WritableTypeInfo任意Hadoop的Writable接口的實現(xiàn)類。
• TupleTypeInfo任意的Flink tuple類型(支持Tuple1 to Tuple25)。 Flink tuples是固定長度固定類型的Java Tuple實現(xiàn)。
• CaseClassTypeInfo任意的 Scala CaseClass(包括 Scala tuples)。
• PojoTypeInfo任意的POJO (Java or Scala)，例如Java對象的所有成員變量，要么是public修飾符定義，要么有g(shù)etter/setter方法。
• GenericTypeInfo任意無法匹配之前幾種類型的類。

前6種類型數(shù)據(jù)集幾乎覆蓋了絕大部分的Flink程序，針對前6種類型數(shù)據(jù)集，F(xiàn)link皆可以自動生成對應的TypeSerializer定制序列化工具，非常有效率地對數(shù)據(jù)集進行序列化和反序列化。對于第7種類型，F(xiàn)link使用Kryo進行序列化和反序列化。此外，對于可被用作Key的類型，F(xiàn)link還同時自動生成TypeComparator，用來輔助直接對序列化后的二進制數(shù)據(jù)直接進行compare、hash等操作。對于Tuple、CaseClass、Pojo等組合類型，F(xiàn)link自動生成的TypeSerializer、TypeComparator同樣是組合的，并把其成員的序列化/反序列化代理給其成員對應的TypeSerializer、TypeComparator，如圖6所示：

此外如有需要，用戶可通過集成TypeInformation接口定制實現(xiàn)自己的序列化工具。

顯式的內(nèi)存管理

垃圾回收是JVM內(nèi)存管理回避不了的問題，JDK8的G1算法改善了JVM垃圾回收的效率和可用范圍，但對于大數(shù)據(jù)處理實際環(huán)境還遠遠不夠。這也和現(xiàn)在分布式框架的發(fā)展趨勢有所沖突，越來越多的分布式計算框架希望盡可能多地將待處理數(shù)據(jù)集放入內(nèi)存，而對于JVM垃圾回收來說，內(nèi)存中Java對象越少、存活時間越短，其效率越高。通過JVM進行內(nèi)存管理的話，OutOfMemoryError也是一個很難解決的問題。同時，在JVM內(nèi)存管理中，Java對象有潛在的碎片化存儲問題(Java對象所有信息可能在內(nèi)存中連續(xù)存儲)，也有可能在所有Java對象大小沒有超過JVM分配內(nèi)存時，出現(xiàn)OutOfMemoryError問題。Flink將內(nèi)存分為3個部分，每個部分都有不同用途：

Network buffers: 一些以32KB Byte數(shù)組為單位的buffer，主要被網(wǎng)絡(luò)模塊用于數(shù)據(jù)的網(wǎng)絡(luò)傳輸。

Memory Manager pool大量以32KB Byte數(shù)組為單位的內(nèi)存池，所有的運行時算法(例如Sort/Shuffle/Join)都從這個內(nèi)存池申請內(nèi)存，并將序列化后的數(shù)據(jù)存儲其中，結(jié)束后釋放回內(nèi)存池。

Remaining (Free) Heap主要留給UDF中用戶自己創(chuàng)建的Java對象，由JVM管理。

Network buffers在Flink中主要基于Netty的網(wǎng)絡(luò)傳輸，無需多講。Remaining Heap用于UDF中用戶自己創(chuàng)建的Java對象，在UDF中，用戶通常是流式的處理數(shù)據(jù)，并不需要很多內(nèi)存，同時Flink也不鼓勵用戶在UDF中緩存很多數(shù)據(jù)，因為這會引起前面提到的諸多問題。Memory Manager pool(以后以內(nèi)存池代指)通常會配置為最大的一塊內(nèi)存，接下來會詳細介紹。

在Flink中，內(nèi)存池由多個MemorySegment組成，每個MemorySegment代表一塊連續(xù)的內(nèi)存，底層存儲是byte[]，默認32KB大小。MemorySegment提供了根據(jù)偏移量訪問數(shù)據(jù)的各種方法，如get/put int、long、float、double等，MemorySegment之間數(shù)據(jù)拷貝等方法和java.nio.ByteBuffer類似。對于Flink的數(shù)據(jù)結(jié)構(gòu)，通常包括多個向內(nèi)存池申請的MemeorySegment，所有要存入的對象通過TypeSerializer序列化之后，將二進制數(shù)據(jù)存儲在MemorySegment中，在取出時通過TypeSerializer反序列化。數(shù)據(jù)結(jié)構(gòu)通過MemorySegment提供的set/get方法訪問具體的二進制數(shù)據(jù)。Flink這種看起來比較復雜的內(nèi)存管理方式帶來的好處主要有：

二進制的數(shù)據(jù)存儲大大提高了數(shù)據(jù)存儲密度，節(jié)省了存儲空間。

所有的運行時數(shù)據(jù)結(jié)構(gòu)和算法只能通過內(nèi)存池申請內(nèi)存，保證了其使用的內(nèi)存大小是固定的，不會因為運行時數(shù)據(jù)結(jié)構(gòu)和算法而發(fā)生OOM。對于大部分的分布式計算框架來說，這部分由于要緩存大量數(shù)據(jù)最有可能導致OOM。

內(nèi)存池雖然占據(jù)了大部分內(nèi)存，但其中的MemorySegment容量較大(默認32KB)，所以內(nèi)存池中的Java對象其實很少，而且一直被內(nèi)存池引用，所有在垃圾回收時很快進入持久代，大大減輕了JVM垃圾回收的壓力。

Remaining Heap的內(nèi)存雖然由JVM管理，但是由于其主要用來存儲用戶處理的流式數(shù)據(jù)，生命周期非常短，速度很快的Minor GC就會全部回收掉，一般不會觸發(fā)Full GC。

Flink當前的內(nèi)存管理在最底層是基于byte[]，所以數(shù)據(jù)最終還是on-heap，最近Flink增加了off-heap的內(nèi)存管理支持。Flink off-heap的內(nèi)存管理相對于on-heap的優(yōu)點主要在于：

啟動分配了大內(nèi)存(例如100G)的JVM很耗費時間，垃圾回收也很慢。如果采用off-heap，剩下的Network buffer和Remaining heap都會很小，垃圾回收也不用考慮MemorySegment中的Java對象了。

更有效率的IO操作。在off-heap下，將MemorySegment寫到磁盤或是網(wǎng)絡(luò)可以支持zeor-copy技術(shù)，而on-heap的話則至少需要一次內(nèi)存拷貝。

off-heap可用于錯誤恢復，比如JVM崩潰，在on-heap時數(shù)據(jù)也隨之丟失，但在off-heap下，off-heap的數(shù)據(jù)可能還在。此外，off-heap上的數(shù)據(jù)還可以和其他程序共享。

緩存友好的計算

磁盤IO和網(wǎng)絡(luò)IO之前一直被認為是Hadoop系統(tǒng)的瓶頸，但是隨著Spark、Flink等新一代分布式計算框架的發(fā)展，越來越多的趨勢使得CPU/Memory逐漸成為瓶頸，這些趨勢包括：

更先進的IO硬件逐漸普及。10GB網(wǎng)絡(luò)和SSD硬盤等已經(jīng)被越來越多的數(shù)據(jù)中心使用。

更高效的存儲格式。Parquet，ORC等列式存儲被越來越多的Hadoop項目支持，其非常高效的壓縮性能大大減少了落地存儲的數(shù)據(jù)量。

更高效的執(zhí)行計劃。例如很多SQL系統(tǒng)執(zhí)行計劃優(yōu)化器的Fliter-Push-Down優(yōu)化會將過濾條件盡可能的提前，甚至提前到Parquet的數(shù)據(jù)訪問層，使得在很多實際的工作負載中并不需要很多的磁盤IO。

由于CPU處理速度和內(nèi)存訪問速度的差距，提升CPU的處理效率的關(guān)鍵在于最大化的利用L1/L2/L3/Memory，減少任何不必要的Cache miss。定制的序列化工具給Flink提供了可能，通過定制的序列化工具，F(xiàn)link訪問的二進制數(shù)據(jù)本身，因為占用內(nèi)存較小，存儲密度比較大，而且還可以在設(shè)計數(shù)據(jù)結(jié)構(gòu)和算法時盡量連續(xù)存儲，減少內(nèi)存碎片化對Cache命中率的影響，甚至更進一步，F(xiàn)link可以只是將需要操作的部分數(shù)據(jù)(如排序時的Key)連續(xù)存儲，而將其他部分的數(shù)據(jù)存儲在其他地方，從而最大可能地提升Cache命中的概率。

以Flink中的排序為例，排序通常是分布式計算框架中一個非常重的操作，F(xiàn)link通過特殊設(shè)計的排序算法獲得了非常好的性能，其排序算法的實現(xiàn)如下：

將待排序的數(shù)據(jù)經(jīng)過序列化后存儲在兩個不同的MemorySegment集中。數(shù)據(jù)全部的序列化值存放于其中一個MemorySegment集中。數(shù)據(jù)序列化后的Key和指向第一個MemorySegment集中值的指針存放于第二個MemorySegment集中。

對第二個MemorySegment集中的Key進行排序，如需交換Key位置，只需交換對應的Key+Pointer的位置，第一個MemorySegment集中的數(shù)據(jù)無需改變。 當比較兩個Key大小時，TypeComparator提供了直接基于二進制數(shù)據(jù)的對比方法，無需反序列化任何數(shù)據(jù)。

排序完成后，訪問數(shù)據(jù)時，按照第二個MemorySegment集中Key的順序訪問，并通過Pointer值找到數(shù)據(jù)在第一個MemorySegment集中的位置，通過TypeSerializer反序列化成Java對象返回。

這樣實現(xiàn)的好處有：

• 通過Key和Full data分離存儲的方式盡量將被操作的數(shù)據(jù)最小化，提高Cache命中的概率，從而提高CPU的吞吐量。
• 移動數(shù)據(jù)時，只需移動Key+Pointer，而無須移動數(shù)據(jù)本身，大大減少了內(nèi)存拷貝的數(shù)據(jù)量。
• TypeComparator直接基于二進制數(shù)據(jù)進行操作，節(jié)省了反序列化的時間。
• 通過定制的內(nèi)存管理，F(xiàn)link通過充分利用內(nèi)存與CPU緩存，大大提高了CPU的執(zhí)行效率，同時由于大部分內(nèi)存都由框架自己控制，也很大程度提升了系統(tǒng)的健壯性，減少了OOM出現(xiàn)的可能。

總結(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項目中去。