相比 Spark Stream、Kafka Stream、Storm 等，為什么阿里會(huì)選擇 Flink 作為新一代流式計(jì)算引擎？前期經(jīng)過(guò)了哪些調(diào)研和對(duì)比？

大沙：我們是 2015 年開(kāi)始調(diào)研新一代流計(jì)算引擎的。我們當(dāng)時(shí)的目標(biāo)就是要設(shè)計(jì)一款低延遲、exactly once、流和批統(tǒng)一的，能夠支撐足夠大體量的復(fù)雜計(jì)算的引擎。Spark streaming 的本質(zhì)還是一款基于 microbatch 計(jì)算的引擎。這種引擎一個(gè)天生的缺點(diǎn)就是每個(gè) microbatch 的調(diào)度開(kāi)銷比較大，當(dāng)我們要求越低的延遲時(shí)，額外的開(kāi)銷就越大。這就導(dǎo)致了 spark streaming 實(shí)際上不是特別適合于做秒級(jí)甚至亞秒級(jí)的計(jì)算。

Kafka streaming 是從一個(gè)日志系統(tǒng)做起來(lái)的，它的設(shè)計(jì)目標(biāo)是足夠輕量，足夠簡(jiǎn)潔易用。這一點(diǎn)很難滿足我們對(duì)大體量的復(fù)雜計(jì)算的需求。

Storm 是一個(gè)沒(méi)有批處理能力的數(shù)據(jù)流處理器，除此之外 Storm 只提供了非常底層的 API，用戶需要自己實(shí)現(xiàn)很多復(fù)雜的邏輯。另外，Storm 在當(dāng)時(shí)不支持 exactly once。種種原因，Storm 也無(wú)法滿足我們的需求。

最后，我們發(fā)現(xiàn)了 Flink，并且驚喜地發(fā)現(xiàn)它幾乎完美滿足了我們所有的需求：

a) 不同于 Spark，Flink 是一個(gè)真正意義上的流計(jì)算引擎，和 Storm 類似，Flink 是通過(guò)流水線數(shù)據(jù)傳輸實(shí)現(xiàn)低延遲的流處理；

b) Flink 使用了經(jīng)典的 Chandy-Lamport 算法，能夠在滿足低延遲和低 failover 開(kāi)銷的基礎(chǔ)之上，完美地解決 exactly once 的目標(biāo)；

c）如果要用一套引擎來(lái)統(tǒng)一流處理和批處理，那就必須以流處理引擎為基礎(chǔ)。Flink 還提供了 SQL／tableAPI 這兩個(gè) API，為批和流在 query 層的統(tǒng)一又鋪平了道路。因此 Flink 是最合適的批和流統(tǒng)一的引擎；

d) 最后，Flink 在設(shè)計(jì)之初就非常在意性能相關(guān)的任務(wù)狀態(tài) state 和流控等關(guān)鍵技術(shù)的設(shè)計(jì)，這些都使得用 Flink 執(zhí)行復(fù)雜的大規(guī)模任務(wù)時(shí)性能更勝一籌。

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

Flink簡(jiǎn)介

Flink核心是一個(gè)流式的數(shù)據(jù)流執(zhí)行引擎,其針對(duì)數(shù)據(jù)流的分布式計(jì)算提供了數(shù)據(jù)分布、數(shù)據(jù)通信以及容錯(cuò)機(jī)制等功能。基于流執(zhí)行引擎,Flink提供了諸多更高抽象層的API以便用戶編寫分布式任務(wù):

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

此外,Flink還針對(duì)特定的應(yīng)用領(lǐng)域提供了領(lǐng)域庫(kù),例如:

Flink ML,Flink的機(jī)器學(xué)習(xí)庫(kù),提供了機(jī)器學(xué)習(xí)Pipelines API并實(shí)現(xiàn)了多種機(jī)器學(xué)習(xí)算法。 Gelly,Flink的圖計(jì)算庫(kù),提供了圖計(jì)算的相關(guān)API及多種圖計(jì)算算法實(shí)現(xiàn)。

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

Flink流計(jì)算隨筆(1)

圖1 Flink技術(shù)棧

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

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

在大數(shù)據(jù)處理領(lǐng)域,批處理任務(wù)與流處理任務(wù)一般被認(rèn)為是兩種不同的任務(wù),一個(gè)大數(shù)據(jù)項(xiàng)目一般會(huì)被設(shè)計(jì)為只能處理其中一種任務(wù),例如Apache Storm、Apache Smaza只支持流處理任務(wù),而Aapche MapReduce、Apache Tez、Apache Spark只支持批處理任務(wù)。Spark Streaming是Apache Spark之上支持流處理任務(wù)的子系統(tǒng),看似一個(gè)特例,實(shí)則不然——Spark Streaming采用了一種micro-batch的架構(gòu),即把輸入的數(shù)據(jù)流切分成細(xì)粒度的batch,并為每一個(gè)batch數(shù)據(jù)提交一個(gè)批處理的Spark任務(wù),所以Spark Streaming本質(zhì)上還是基于Spark批處理系統(tǒng)對(duì)流式數(shù)據(jù)進(jìn)行處理,和Apache Storm、Apache Smaza等完全流式的數(shù)據(jù)處理方式完全不同。通過(guò)其靈活的執(zhí)行引擎,Flink能夠同時(shí)支持批處理任務(wù)與流處理任務(wù)。

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

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

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

Flink流處理的容錯(cuò)機(jī)制

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

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

按照用戶自定義的分布式快照間隔時(shí)間,Flink會(huì)定時(shí)在所有數(shù)據(jù)源中插入一種特殊的快照標(biāo)記消息,這些快照標(biāo)記消息和其他消息一樣在DAG中流動(dòng),但是不會(huì)被用戶定義的業(yè)務(wù)邏輯所處理,每一個(gè)快照標(biāo)記消息都將其所在的數(shù)據(jù)流分成兩部分:本次快照數(shù)據(jù)和下次快照數(shù)據(jù)。

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

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

圖4 Flink分布式快照流程圖

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

Exactly-Once是流處理系統(tǒng)需要支持的一個(gè)非常重要的特性,它保證每一條消息只被流處理系統(tǒng)處理一次,許多流處理任務(wù)的業(yè)務(wù)邏輯都依賴于Exactly-Once特性。相對(duì)于At-Least-Once或是At-Most-Once, Exactly-Once特性對(duì)流處理系統(tǒng)的要求更為嚴(yán)格,實(shí)現(xiàn)也更加困難。Flink基于分布式快照實(shí)現(xiàn)了Exactly-Once特性。

相對(duì)于其他流處理系統(tǒng)的容錯(cuò)方案,Flink基于分布式快照的方案在功能和性能方面都具有很多優(yōu)點(diǎn),包括:

低延遲。由于操作符狀態(tài)的存儲(chǔ)可以異步,所以進(jìn)行快照的過(guò)程基本上不會(huì)阻塞消息的處理,因此不會(huì)對(duì)消息延遲產(chǎn)生負(fù)面影響。 高吞吐量。當(dāng)操作符狀態(tài)較少時(shí),對(duì)吞吐量基本沒(méi)有影響。當(dāng)操作符狀態(tài)較多時(shí),相對(duì)于其他的容錯(cuò)機(jī)制,分布式快照的時(shí)間間隔是用戶自定義的,所以用戶可以權(quán)衡錯(cuò)誤恢復(fù)時(shí)間和吞吐量要求來(lái)調(diào)整分布式快照的時(shí)間間隔。 與業(yè)務(wù)邏輯的隔離。Flink的分布式快照機(jī)制與用戶的業(yè)務(wù)邏輯是完全隔離的,用戶的業(yè)務(wù)邏輯不會(huì)依賴或是對(duì)分布式快照產(chǎn)生任何影響。 錯(cuò)誤恢復(fù)代價(jià)。分布式快照的時(shí)間間隔越短,錯(cuò)誤恢復(fù)的時(shí)間越少,與吞吐量負(fù)相關(guān)。

Flink流處理的時(shí)間窗口

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

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

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

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

Flink支持3種類型的時(shí)間窗口,分別適用于用戶對(duì)于時(shí)間窗口不同類型的要求:

Operator Time。根據(jù)Task所在節(jié)點(diǎn)的本地時(shí)鐘來(lái)切分的時(shí)間窗口。

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

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

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

Flink借鑒了Google的MillWheel項(xiàng)目,通過(guò)WaterMark來(lái)支持基于Event Time的時(shí)間窗口。

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

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

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

圖5 WaterMark的Time Skew圖

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

基于時(shí)間戳的排序

在流處理系統(tǒng)中,由于流入的消息是無(wú)限的,所以對(duì)消息進(jìn)行排序基本上被認(rèn)為是不可行的。但是在Flink流處理系統(tǒng)中,基于WaterMark,Flink實(shí)現(xiàn)了基于時(shí)間戳的全局排序。排序的實(shí)現(xiàn)思路如下:排序操作符緩存所有流入的消息,當(dāng)其接收到WaterMark時(shí),對(duì)時(shí)間戳小于該WaterMark的消息進(jìn)行排序,并發(fā)送到下一個(gè)節(jié)點(diǎn),在此排序操作符中釋放所有時(shí)間戳小于該WaterMark的消息,繼續(xù)緩存流入的消息,等待下一個(gè)WaterMark觸發(fā)下一次排序。

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

通過(guò)支持基于Event Time的消息處理,Flink擴(kuò)展了其流處理系統(tǒng)的應(yīng)用范圍,使得更多的流處理任務(wù)可以通過(guò)Flink來(lái)執(zhí)行。

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

Flink項(xiàng)目基于Java及Scala等JVM語(yǔ)言,JVM本身作為一個(gè)各種類型應(yīng)用的執(zhí)行平臺(tái),其對(duì)Java對(duì)象的管理也是基于通用的處理策略,其垃圾回收器通過(guò)估算Java對(duì)象的生命周期對(duì)Java對(duì)象進(jìn)行有效率的管理。

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

JVM存在的問(wèn)題

Java對(duì)象開(kāi)銷

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

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

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

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

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

OOM問(wèn)題

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

Flink的處理策略

為了解決以上提到的問(wèn)題,高性能分布式計(jì)算框架通常需要以下技術(shù):

定制的序列化工具。顯式內(nèi)存管理的前提步驟就是序列化,將Java對(duì)象序列化成二進(jìn)制數(shù)據(jù)存儲(chǔ)在內(nèi)存上(on heap或是off-heap)。通用的序列化框架,如Java默認(rèn)使用java.io.Serializable將Java對(duì)象及其成員變量的所有元信息作為其序列化數(shù)據(jù)的一部分,序列化后的數(shù)據(jù)包含了所有反序列化所需的信息。這在某些場(chǎng)景中十分必要,但是對(duì)于Flink這樣的分布式計(jì)算框架來(lái)說(shuō),這些元數(shù)據(jù)信息可能是冗余數(shù)據(jù)。定制的序列化框架,如Hadoop的org.apache.hadoop.io.Writable需要用戶實(shí)現(xiàn)該接口,并自定義類的序列化和反序列化方法。這種方式效率最高,但需要用戶額外的工作,不夠友好。 顯式的內(nèi)存管理。一般通用的做法是批量申請(qǐng)和釋放內(nèi)存,每個(gè)JVM實(shí)例有一個(gè)統(tǒng)一的內(nèi)存管理器,所有內(nèi)存的申請(qǐng)和釋放都通過(guò)該內(nèi)存管理器進(jìn)行。這可以避免常見(jiàn)的內(nèi)存碎片問(wèn)題,同時(shí)由于數(shù)據(jù)以二進(jìn)制的方式存儲(chǔ),可以大大減輕垃圾回收壓力。

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

定制的序列化工具

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

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

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

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

圖6 Flink組合類型序列化

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

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

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

Network buffers: 一些以32KB Byte數(shù)組為單位的buffer,主要被網(wǎng)絡(luò)模塊用于數(shù)據(jù)的網(wǎng)絡(luò)傳輸。 Memory Manager pool大量以32KB Byte數(shù)組為單位的內(nèi)存池,所有的運(yùn)行時(shí)算法(例如Sort/Shuffle/Join)都從這個(gè)內(nèi)存池申請(qǐng)內(nèi)存,并將序列化后的數(shù)據(jù)存儲(chǔ)其中,結(jié)束后釋放回內(nèi)存池。 Remaining (Free) Heap主要留給UDF中用戶自己創(chuàng)建的Java對(duì)象,由JVM管理。

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

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

二進(jìn)制的數(shù)據(jù)存儲(chǔ)大大提高了數(shù)據(jù)存儲(chǔ)密度,節(jié)省了存儲(chǔ)空間。 所有的運(yùn)行時(shí)數(shù)據(jù)結(jié)構(gòu)和算法只能通過(guò)內(nèi)存池申請(qǐng)內(nèi)存,保證了其使用的內(nèi)存大小是固定的,不會(huì)因?yàn)檫\(yùn)行時(shí)數(shù)據(jù)結(jié)構(gòu)和算法而發(fā)生OOM。對(duì)于大部分的分布式計(jì)算框架來(lái)說(shuō),這部分由于要緩存大量數(shù)據(jù)最有可能導(dǎo)致OOM。 內(nèi)存池雖然占據(jù)了大部分內(nèi)存,但其中的MemorySegment容量較大(默認(rèn)32KB),所以內(nèi)存池中的Java對(duì)象其實(shí)很少,而且一直被內(nèi)存池引用,所有在垃圾回收時(shí)很快進(jìn)入持久代,大大減輕了JVM垃圾回收的壓力。 Remaining Heap的內(nèi)存雖然由JVM管理,但是由于其主要用來(lái)存儲(chǔ)用戶處理的流式數(shù)據(jù),生命周期非常短,速度很快的Minor GC就會(huì)全部回收掉,一般不會(huì)觸發(fā)Full GC。

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

啟動(dòng)分配了大內(nèi)存(例如100G)的JVM很耗費(fèi)時(shí)間,垃圾回收也很慢。如果采用off-heap,剩下的Network buffer和Remaining heap都會(huì)很小,垃圾回收也不用考慮MemorySegment中的Java對(duì)象了。 更有效率的IO操作。在off-heap下,將MemorySegment寫到磁盤或是網(wǎng)絡(luò)可以支持zeor-copy技術(shù),而on-heap的話則至少需要一次內(nèi)存拷貝。 off-heap可用于錯(cuò)誤恢復(fù),比如JVM崩潰,在on-heap時(shí)數(shù)據(jù)也隨之丟失,但在off-heap下,off-heap的數(shù)據(jù)可能還在。此外,off-heap上的數(shù)據(jù)還可以和其他程序共享。 緩存友好的計(jì)算

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

更先進(jìn)的IO硬件逐漸普及。10GB網(wǎng)絡(luò)和SSD硬盤等已經(jīng)被越來(lái)越多的數(shù)據(jù)中心使用。 更高效的存儲(chǔ)格式。Parquet,ORC等列式存儲(chǔ)被越來(lái)越多的Hadoop項(xiàng)目支持,其非常高效的壓縮性能大大減少了落地存儲(chǔ)的數(shù)據(jù)量。 更高效的執(zhí)行計(jì)劃。例如很多SQL系統(tǒng)執(zhí)行計(jì)劃優(yōu)化器的Fliter-Push-Down優(yōu)化會(huì)將過(guò)濾條件盡可能的提前,甚至提前到Parquet的數(shù)據(jù)訪問(wèn)層,使得在很多實(shí)際的工作負(fù)載中并不需要很多的磁盤IO。

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

以Flink中的排序?yàn)槔?排序通常是分布式計(jì)算框架中一個(gè)非常重的操作,Flink通過(guò)特殊設(shè)計(jì)的排序算法獲得了非常好的性能,其排序算法的實(shí)現(xiàn)如下:

將待排序的數(shù)據(jù)經(jīng)過(guò)序列化后存儲(chǔ)在兩個(gè)不同的MemorySegment集中。數(shù)據(jù)全部的序列化值存放于其中一個(gè)MemorySegment集中。數(shù)據(jù)序列化后的Key和指向第一個(gè)MemorySegment集中值的指針存放于第二個(gè)MemorySegment集中。 對(duì)第二個(gè)MemorySegment集中的Key進(jìn)行排序,如需交換Key位置,只需交換對(duì)應(yīng)的Key+Pointer的位置,第一個(gè)MemorySegment集中的數(shù)據(jù)無(wú)需改變。 當(dāng)比較兩個(gè)Key大小時(shí),TypeComparator提供了直接基于二進(jìn)制數(shù)據(jù)的對(duì)比方法,無(wú)需反序列化任何數(shù)據(jù)。 排序完成后,訪問(wèn)數(shù)據(jù)時(shí),按照第二個(gè)MemorySegment集中Key的順序訪問(wèn),并通過(guò)Pointer值找到數(shù)據(jù)在第一個(gè)MemorySegment集中的位置,通過(guò)TypeSerializer反序列化成Java對(duì)象返回。

圖7 Flink排序算法

這樣實(shí)現(xiàn)的好處有:

通過(guò)Key和Full data分離存儲(chǔ)的方式盡量將被操作的數(shù)據(jù)最小化,提高Cache命中的概率,從而提高CPU的吞吐量。 移動(dòng)數(shù)據(jù)時(shí),只需移動(dòng)Key+Pointer,而無(wú)須移動(dòng)數(shù)據(jù)本身,大大減少了內(nèi)存拷貝的數(shù)據(jù)量。 TypeComparator直接基于二進(jìn)制數(shù)據(jù)進(jìn)行操作,節(jié)省了反序列化的時(shí)間。

通過(guò)定制的內(nèi)存管理,Flink通過(guò)充分利用內(nèi)存與CPU緩存,大大提高了CPU的執(zhí)行效率,同時(shí)由于大部分內(nèi)存都由框架自己控制,也很大程度提升了系統(tǒng)的健壯性,減少了OOM出現(xiàn)的可能。

MACOS下安裝flink:

$ brew install apache-flink ... $ flink --version $brew upgrade

MACOS下啟動(dòng)flink:

$cd /usr/local/Cellar/apache-flink/1.6.0$./libexec/bin/start-cluster.sh /** Licensed to the Apache Software Foundation (ASF) under one* or more contributor license agreements. See the NOTICE file* distributed with this work for additional information* regarding copyright ownership. The ASF licenses this file* to you under the Apache License, Version 2.0 (the* "License"); you may not use this file except in compliance* with the License. You may obtain a copy of the License at** http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/package org.apache.flink.streaming.scala.examples.socketimport org.apache.flink.api.java.utils.ParameterTool import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.api.windowing.time.Time/*** Implements a streaming windowed version of the "WordCount" program.* * This program connects to a server socket and reads strings from the socket.* The easiest way to try this out is to open a text sever (at port 12345) * using the ''netcat'' tool via* {{{* nc -l 12345* }}}* and run this example with the hostname and the port as arguments..*/ object SocketWindowWordCount {/** Main program method */def main(args: Array[String]) : Unit = {// the host and the port to connect tovar hostname: String = "localhost"var port: Int = 0try {val params = ParameterTool.fromArgs(args)hostname = if (params.has("hostname")) params.get("hostname") else "localhost"port = params.getInt("port")} catch {case e: Exception => {System.err.println("No port specified. Please run 'SocketWindowWordCount " +"--hostname <hostname> --port <port>', where hostname (localhost by default) and port " +"is the address of the text server")System.err.println("To start a simple text server, run 'netcat -l <port>' " +"and type the input text into the command line")return}}// get the execution environmentval env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment// get input data by connecting to the socketval text: DataStream[String] = env.socketTextStream(hostname, port, '\n')// parse the data, group it, window it, and aggregate the counts val windowCounts = text.flatMap { w => w.split("\\s") }.map { w => WordWithCount(w, 1) }.keyBy("word").timeWindow(Time.seconds(5)).sum("count")// print the results with a single thread, rather than in parallelwindowCounts.print().setParallelism(1)env.execute("Socket Window WordCount")}/** Data type for words with count */case class WordWithCount(word: String, count: Long) }

訪問(wèn)127.0.0.1:8081

測(cè)試:

//執(zhí)行完nc輸入單詞，程序會(huì)開(kāi)始記數(shù)。$nc -l 9002//開(kāi)另一個(gè)xshell，執(zhí)行運(yùn)行程序的命令 $flink run /usr/local/Cellar/apache-flink/1.6.0/libexec/examples/streaming/SocketWindowWordCount.jar --port 9002 //到log目錄下可以看到輸出了記數(shù)的文件

linux下安裝

1.Download a binary from the downloads page2.$ cd ~/Downloads # Go to download directory $ tar xzf flink-*.tgz # Unpack the downloaded archive $ cd flink-1.6.1 3.$ ./bin/start-cluster.sh # Start Flinkhttp://localhost:8081 訪問(wèn)bogon:1.6.0 myhaspl$ ./libexec/bin/stop-cluster.shStopping taskexecutor daemon (pid: 11100) on host bogon.Stopping standalonesession daemon (pid: 10689) on host bogon.

Stateful Computations over Data Streams(在數(shù)據(jù)流的有狀態(tài)計(jì)算)
Apache Flink是一個(gè)用于分布式流和批處理數(shù)據(jù)的開(kāi)源平臺(tái)。Flink的核心是一個(gè)流數(shù)據(jù)流引擎，它為數(shù)據(jù)流上的分布式計(jì)算提供數(shù)據(jù)分布、通信和容錯(cuò)能力。Flink在流引擎之上構(gòu)建批處理，覆蓋本地迭代支持、托管內(nèi)存和程序優(yōu)化。
通常在程序中的轉(zhuǎn)換和數(shù)據(jù)流中的操作符之間存在一對(duì)一的對(duì)應(yīng)關(guān)系。然而，有時(shí)一個(gè)轉(zhuǎn)換可能包含多個(gè)轉(zhuǎn)換操作符。

在串流連接器和批處理連接器文檔中記錄了源和匯（Sources and sinks）。在DataStream運(yùn)算符和數(shù)據(jù)集轉(zhuǎn)換中記錄了轉(zhuǎn)換。

Flink提供了不同級(jí)別的抽象來(lái)開(kāi)發(fā)流/批處理應(yīng)用程序。
flink流計(jì)算隨筆(3)

最低層次的抽象僅僅提供有狀態(tài)流（ stateful streaming）。它通過(guò)Process函數(shù)嵌入到DataStream API中。它允許用戶自由地處理來(lái)自一個(gè)或多個(gè)流的事件，并使用一致的容錯(cuò)狀態(tài)。此外，用戶可以注冊(cè)事件時(shí)間和處理時(shí)間回調(diào)，允許程序?qū)崿F(xiàn)復(fù)雜的計(jì)算。

在實(shí)踐中，大多數(shù)應(yīng)用程序不需要上面描述的低級(jí)抽象，而是對(duì)核心API（Core APIs ）進(jìn)行編程，比如DataStream API(有界/×××流)和DataSet API(有界數(shù)據(jù)集)。這些fluent api提供了用于數(shù)據(jù)處理的通用構(gòu)建塊，比如各種形式的用戶指定的轉(zhuǎn)換、連接、聚合、窗口、狀態(tài)等。在這些api中處理的數(shù)據(jù)類型表示為各自編程語(yǔ)言中的類。

低級(jí)流程函數(shù)與DataStream API集成，使得只對(duì)某些操作進(jìn)行低級(jí)抽象成為可能。DataSet API為有界數(shù)據(jù)集提供了額外的原語(yǔ)，比如循環(huán)/迭代。

表API是一個(gè)以表為中心的聲明性DSL，它可以動(dòng)態(tài)地改變表(當(dāng)表示流時(shí))。表API遵循(擴(kuò)展)關(guān)系模型:表有一個(gè)附加模式(類似于關(guān)系數(shù)據(jù)庫(kù)表)和API提供了類似的操作,如select, project, join, group-by, aggregate等。表API程序以聲明的方式定義邏輯操作應(yīng)該做什么而不是指定操作的代碼看起來(lái)如何。雖然表API可以通過(guò)各種用戶定義函數(shù)進(jìn)行擴(kuò)展，但它的表達(dá)性不如核心API，但使用起來(lái)更簡(jiǎn)潔(編寫的代碼更少)。此外，表API程序還通過(guò)一個(gè)優(yōu)化器在執(zhí)行之前應(yīng)用優(yōu)化規(guī)則。

可以無(wú)縫地在表和DataStream/DataSet之間進(jìn)行轉(zhuǎn)換，允許程序混合表API和DataStream和DataSet API。

Flink提供的最高級(jí)別抽象是SQL。這種抽象在語(yǔ)義和表示方面都類似于表API，但將程序表示為SQL查詢表達(dá)式。SQL抽象與表API密切交互，SQL查詢可以在表API中定義的表上執(zhí)行。
Flink程序的基本構(gòu)建模塊是流和轉(zhuǎn)換(streams and transformations)。(請(qǐng)注意，Flink的DataSet API中使用的數(shù)據(jù)集也是內(nèi)部流。)從概念上講，流是數(shù)據(jù)記錄的(可能是無(wú)限的)流，而轉(zhuǎn)換是將一個(gè)或多個(gè)流作為輸入并產(chǎn)生一個(gè)或多個(gè)輸出流的操作。

執(zhí)行時(shí)，Flink程序被映射到流數(shù)據(jù)流streaming dataflows,，由流和轉(zhuǎn)換操作符組成。每個(gè)數(shù)據(jù)流以一個(gè)或多個(gè)源開(kāi)始，以一個(gè)或多個(gè)接收器結(jié)束。數(shù)據(jù)流類似于任意有向無(wú)環(huán)圖(DAGs)。雖然通過(guò)迭代構(gòu)造允許特殊形式的循環(huán)，但為了簡(jiǎn)單起見(jiàn)將在大多數(shù)情況下忽略這一點(diǎn)。

總結(jié)

以上是生活随笔為你收集整理的Flink随笔(1）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。