戳藍(lán)字“CSDN云計(jì)算”關(guān)注我們哦！

文?|??張馨予??

? 來(lái)源?| 高可用架構(gòu)

作者?張馨予，360 大數(shù)據(jù)計(jì)算平臺(tái)負(fù)責(zé)人。北京郵電大學(xué)碩士，2015年加入360系統(tǒng)部，一直致力于公司大數(shù)據(jù)計(jì)算平臺(tái)的易用性、穩(wěn)定性和性能優(yōu)化的研發(fā)工作。目前主要負(fù)責(zé)Flink的研發(fā)，完成公司計(jì)算引擎的大一統(tǒng)。

本文從數(shù)據(jù)傳輸和數(shù)據(jù)可靠性的角度出發(fā)，對(duì)比測(cè)試了Storm與Flink在流處理上的性能，并對(duì)測(cè)試結(jié)果進(jìn)行分析，給出在使用Flink時(shí)提高性能的建議。

Apache Storm、Apache Spark和Apache Flink都是開(kāi)源社區(qū)中非常活躍的分布式計(jì)算平臺(tái)，在很多公司可能同時(shí)使用著其中兩種甚至三種。對(duì)于實(shí)時(shí)計(jì)算來(lái)說(shuō)，Storm與Flink的底層計(jì)算引擎是基于流的，本質(zhì)上是一條一條的數(shù)據(jù)進(jìn)行處理，且處理的模式是流水線模式，即所有的處理進(jìn)程同時(shí)存在，數(shù)據(jù)在這些進(jìn)程之間流動(dòng)處理。而Spark是基于批量數(shù)據(jù)的處理，即一小批一小批的數(shù)據(jù)進(jìn)行處理，且處理的邏輯在一批數(shù)據(jù)準(zhǔn)備好之后才會(huì)進(jìn)行計(jì)算。在本文中，我們把同樣基于流處理的Storm和Flink拿來(lái)做對(duì)比測(cè)試分析。

在我們做測(cè)試之前，調(diào)研了一些已有的大數(shù)據(jù)平臺(tái)性能測(cè)試報(bào)告，比如，雅虎的Streaming-benchmarks，或者Intel的HiBench等等。除此之外，還有很多的論文也從不同的角度對(duì)分布式計(jì)算平臺(tái)進(jìn)行了測(cè)試。雖然這些測(cè)試case各有不同的側(cè)重點(diǎn)，但他們都用到了同樣的兩個(gè)指標(biāo)，即吞吐和延遲。吞吐表示單位時(shí)間內(nèi)所能處理的數(shù)據(jù)量，是可以通過(guò)增大并發(fā)來(lái)提高的。延遲代表處理一條數(shù)據(jù)所需要的時(shí)間，與吞吐量成反比關(guān)系。

在我們?cè)O(shè)計(jì)計(jì)算邏輯時(shí)，首先考慮一下流處理的計(jì)算模型。上圖是一個(gè)簡(jiǎn)單的流計(jì)算模型，在Source中將數(shù)據(jù)取出，發(fā)往下游Task，并在Task中進(jìn)行處理，最后輸出。對(duì)于這樣的一個(gè)計(jì)算模型，延遲時(shí)間由三部分組成：數(shù)據(jù)傳輸時(shí)間、Task計(jì)算時(shí)間和數(shù)據(jù)排隊(duì)時(shí)間。我們假設(shè)資源足夠，數(shù)據(jù)不用排隊(duì)。則延遲時(shí)間就只由數(shù)據(jù)傳輸時(shí)間和Task計(jì)算時(shí)間組成。而在Task中處理所需要的時(shí)間與用戶的邏輯息息相關(guān)，所以對(duì)于一個(gè)計(jì)算平臺(tái)來(lái)說(shuō)，數(shù)據(jù)傳輸?shù)臅r(shí)間才更能反映這個(gè)計(jì)算平臺(tái)的能力。因此，我們?cè)谠O(shè)計(jì)測(cè)試Case時(shí)，為了更好的體現(xiàn)出數(shù)據(jù)傳輸?shù)哪芰?#xff0c;Task中沒(méi)有設(shè)計(jì)任何計(jì)算邏輯。

在確定數(shù)據(jù)源時(shí)，我們主要考慮是在進(jìn)程中直接生成數(shù)據(jù)，這種方法在很多之前的測(cè)試標(biāo)準(zhǔn)中也同樣有使用。這樣做是因?yàn)閿?shù)據(jù)的產(chǎn)生不會(huì)受到外界數(shù)據(jù)源系統(tǒng)的性能限制。但由于在我們公司內(nèi)部大部分的實(shí)時(shí)計(jì)算數(shù)據(jù)都來(lái)源于kafka，所以我們?cè)黾恿藦膋afka中讀取數(shù)據(jù)的測(cè)試。

對(duì)于數(shù)據(jù)傳輸方式，可以分為兩種：進(jìn)程間的數(shù)據(jù)傳輸和進(jìn)程內(nèi)的數(shù)據(jù)傳輸。

進(jìn)程間的數(shù)據(jù)傳輸是指這條數(shù)據(jù)會(huì)經(jīng)過(guò)序列化、網(wǎng)絡(luò)傳輸和反序列化三個(gè)步驟。在Flink中，2個(gè)處理邏輯分布在不同的TaskManager上，這兩個(gè)處理邏輯之間的數(shù)據(jù)傳輸就可以叫做進(jìn)程間的數(shù)據(jù)傳輸。Flink網(wǎng)絡(luò)傳輸是采用的Netty技術(shù)。在Storm中，進(jìn)程間的數(shù)據(jù)傳輸是worker之間的數(shù)據(jù)傳輸。早版本的storm網(wǎng)絡(luò)傳輸使用的ZeroMQ，現(xiàn)在也改成了Netty。

進(jìn)程內(nèi)的數(shù)據(jù)傳輸是指兩個(gè)處理邏輯在同一個(gè)進(jìn)程中。在Flink中，這兩個(gè)處理邏輯被Chain在了一起，在一個(gè)線程中通過(guò)方法調(diào)用傳參的形式進(jìn)程數(shù)據(jù)傳輸。在Storm中，兩個(gè)處理邏輯變成了兩個(gè)線程，通過(guò)一個(gè)共享的隊(duì)列進(jìn)行數(shù)據(jù)傳輸。

Storm和Flink都有各自的可靠性機(jī)制。在Storm中，使用ACK機(jī)制來(lái)保證數(shù)據(jù)的可靠性。而在Flink中是通過(guò)checkpoint機(jī)制來(lái)保證的，這是來(lái)源于chandy-lamport算法。

事實(shí)上exactly-once可靠性的保證跟處理的邏輯和結(jié)果輸出的設(shè)計(jì)有關(guān)。比如結(jié)果要輸出到kafka中，而輸出到kafka的數(shù)據(jù)無(wú)法回滾，這就無(wú)法保證exactly-once。我們?cè)跍y(cè)試的時(shí)候選用的at-least-once語(yǔ)義的可靠性和不保證可靠性兩種策略進(jìn)行測(cè)試。

上圖是我們測(cè)試的環(huán)境和各個(gè)平臺(tái)的版本。

上圖展示的是Flink在自產(chǎn)數(shù)據(jù)的情況下，不同的傳輸方式和可靠性的吞吐量：在進(jìn)程內(nèi)+不可靠、進(jìn)程內(nèi)+可靠、進(jìn)程間+不可靠、進(jìn)程間+可靠。可以看到進(jìn)程內(nèi)的數(shù)據(jù)傳輸是進(jìn)程間的數(shù)據(jù)傳輸?shù)?.8倍。是否開(kāi)啟checkpoint機(jī)制對(duì)Flink的吞吐影響并不大。因此我們?cè)谑褂肍link時(shí)，進(jìn)來(lái)使用進(jìn)程內(nèi)的傳輸，也就是盡可能的讓算子可以Chain起來(lái)。

那么我們來(lái)看一下為什么Chain起來(lái)的性能好這么多，要如何在寫(xiě)Flink代碼的過(guò)程中讓Flink的算子Chain起來(lái)使用進(jìn)程間的數(shù)據(jù)傳輸。

大家知道我們?cè)贔link代碼時(shí)一定會(huì)創(chuàng)建一個(gè)env，調(diào)用env的disableOperatorChainning()方法會(huì)使得所有的算子都無(wú)法chain起來(lái)。我們一般是在debug的時(shí)候回調(diào)用這個(gè)方法，方便調(diào)試問(wèn)題。

如果允許Chain的情況下，上圖中Source和mapFunction就會(huì)Chain起來(lái)，放在一個(gè)Task中計(jì)算。反之，如果不允許Chain，則會(huì)放到兩個(gè)Task中。

對(duì)于沒(méi)有Chain起來(lái)的兩個(gè)算子，他們被放到了不同的兩個(gè)Task中，那么他們之間的數(shù)據(jù)傳輸是這樣的：SourceFunction取到數(shù)據(jù)序列化后放入內(nèi)存，然后通過(guò)網(wǎng)絡(luò)傳輸給MapFunction所在的進(jìn)程，該進(jìn)程將數(shù)據(jù)方序列化后使用。

對(duì)于Chain起來(lái)的兩個(gè)算子，他們被放到同一個(gè)Task中，那么這兩個(gè)算子之間的數(shù)據(jù)傳輸則是：SourceFunction取到數(shù)據(jù)后，進(jìn)行一次深拷貝，然后MapFunction把深拷貝出來(lái)的這個(gè)對(duì)象作為輸入數(shù)據(jù)。

雖然Flink在序列化上做了很多優(yōu)化，跟不用序列化和不用網(wǎng)絡(luò)傳輸?shù)倪M(jìn)程內(nèi)數(shù)據(jù)傳輸對(duì)比，性能還是差很多。所以我們盡可能的把算子Chain起來(lái)。

不是任何兩個(gè)算子都可以Chain起來(lái)的，要把算子Chain起來(lái)有很多條件：第一，下游算子只能接受一種上游數(shù)據(jù)流，比如Map接受的流不能是一條union后的流；其次上下游的并發(fā)數(shù)一定要一樣；第三，算子要使用同一個(gè)資源Group，默認(rèn)是一致的，都是default；第四，就是之前說(shuō)的env中不能調(diào)用disableOperatorChainning()方法，最后，上游發(fā)送數(shù)據(jù)的方法是Forward的，比如，開(kāi)發(fā)時(shí)沒(méi)有調(diào)用rebalance()方法，沒(méi)有keyby()，沒(méi)有boardcast等。

對(duì)比一下自產(chǎn)數(shù)據(jù)時(shí)，使用進(jìn)程內(nèi)通信，且不保證數(shù)據(jù)可靠性的情況下，Flink與Storm的吞吐。在這種情況下，Flink的性能是Storm的15倍。Flink吞吐能達(dá)到2060萬(wàn)條/s。不僅如此，如果在開(kāi)發(fā)時(shí)調(diào)用了env.getConfig().enableObjectReuse()方法，Flink的但并發(fā)吞吐能達(dá)到4090萬(wàn)條/s。

當(dāng)調(diào)用了enableObjectReuse方法后，Flink會(huì)把中間深拷貝的步驟都省略掉，SourceFunction產(chǎn)生的數(shù)據(jù)直接作為MapFunction的輸入。但需要特別注意的是，這個(gè)方法不能隨便調(diào)用，必須要確保下游Function只有一種，或者下游的Function均不會(huì)改變對(duì)象內(nèi)部的值。否則可能會(huì)有線程安全的問(wèn)題。

當(dāng)對(duì)比在不同可靠性策略的情況下，Flink與Storm的表現(xiàn)時(shí)，我們發(fā)現(xiàn)，保證可靠性對(duì)Flink的影響非常小，但對(duì)Storm的影響非常大。總的來(lái)說(shuō)，在保證可靠的情況下，Flink單并發(fā)的吞吐是Storm的15倍，而不保證可靠的情況下，Flink的性能是Storm的66倍。會(huì)產(chǎn)生這樣的結(jié)果，主要是因?yàn)镕link與Storm保證數(shù)據(jù)可靠性的機(jī)制不同。

而Storm的ACK機(jī)制為了保證數(shù)據(jù)的可靠性，開(kāi)銷更大。

左邊的圖展示的是Storm的Ack機(jī)制。Spout每發(fā)送一條數(shù)據(jù)到Bolt，就會(huì)產(chǎn)生一條ack的信息給acker，當(dāng)Bolt處理完這條數(shù)據(jù)后也會(huì)發(fā)送ack信息給acker。當(dāng)acker收到這條數(shù)據(jù)的所有ack信息時(shí)，會(huì)回復(fù)Spout一條ack信息。也就是說(shuō)，對(duì)于一個(gè)只有兩級(jí)（spout+bolt）的拓?fù)鋪?lái)說(shuō)，每發(fā)送一條數(shù)據(jù)，就會(huì)傳輸3條ack信息。這3條ack信息則是為了保證可靠性所需要的開(kāi)銷。

右邊的圖展示的是Flink的Checkpoint機(jī)制。Flink中Checkpoint信息的發(fā)起者是JobManager。它不像Storm中那樣，每條信息都會(huì)有ack信息的開(kāi)銷，而且按時(shí)間來(lái)計(jì)算花銷。用戶可以設(shè)置做checkpoint的頻率，比如10秒鐘做一次checkpoint。每做一次checkpoint，花銷只有從Source發(fā)往map的1條checkpoint信息（JobManager發(fā)出來(lái)的checkpoint信息走的是控制流，與數(shù)據(jù)流無(wú)關(guān)）。與storm相比，Flink的可靠性機(jī)制開(kāi)銷要低得多。這也就是為什么保證可靠性對(duì)Flink的性能影響較小，而storm的影響確很大的原因。

最后一組自產(chǎn)數(shù)據(jù)的測(cè)試結(jié)果對(duì)比是Flink與Storm在進(jìn)程間的數(shù)據(jù)傳輸?shù)膶?duì)比，可以看到進(jìn)程間數(shù)據(jù)傳輸?shù)那闆r下，Flink但并發(fā)吞吐是Storm的4.7倍。保證可靠性的情況下，是Storm的14倍。

上圖展示的是消費(fèi)kafka中數(shù)據(jù)時(shí)，Storm與Flink的但并發(fā)吞吐情況。因?yàn)橄M(fèi)的是kafka中的數(shù)據(jù)，所以吞吐量肯定會(huì)收到kafka的影響。我們發(fā)現(xiàn)性能的瓶頸是在SourceFunction上，于是增加了topic的partition數(shù)和SourceFunction取數(shù)據(jù)線程的并發(fā)數(shù)，但是MapFunction的并發(fā)數(shù)仍然是1.在這種情況下，我們發(fā)現(xiàn)flink的瓶頸轉(zhuǎn)移到上游往下游發(fā)數(shù)據(jù)的地方。而Storm的瓶頸確是在下游收數(shù)據(jù)反序列化的地方。

之前的性能分析使我們基于數(shù)據(jù)傳輸和數(shù)據(jù)可靠性的角度出發(fā)，單純的對(duì)Flink與Storm計(jì)算平臺(tái)本身進(jìn)行了性能分析。但實(shí)際使用時(shí)，task是肯定有計(jì)算邏輯的，這就勢(shì)必更多的涉及到CPU，內(nèi)存等資源問(wèn)題。我們將來(lái)打算做一個(gè)智能分析平臺(tái)，對(duì)用戶的作業(yè)進(jìn)行性能分析。通過(guò)收集到的指標(biāo)信息，分析出作業(yè)的瓶頸在哪，并給出優(yōu)化建議。

福利

掃描添加小編微信，備注“姓名+公司職位”，加入【云計(jì)算學(xué)習(xí)交流群】，和志同道合的朋友們共同打卡學(xué)習(xí)！

推薦閱讀：

• IEEE 回應(yīng)禁止華為系審稿人;WiFi聯(lián)盟、藍(lán)牙聯(lián)盟已恢復(fù)華為成員資格；中國(guó)計(jì)算機(jī)學(xué)會(huì)：暫時(shí)中止與IEEE通信學(xué)會(huì)合作……

• ARM 發(fā)布新一代 CPU 和 GPU，實(shí)現(xiàn) 20% 性能提升！

• 前端開(kāi)發(fā) 20 年變遷史
  

• 北漂杭漂的程序員，是如何買(mǎi)到第一套房子？
  

• “愛(ài)裝X”開(kāi)源組織：“教科書(shū)級(jí)”AI知識(shí)樹(shù)究竟長(zhǎng)什么樣？

• 500行Python代碼打造刷臉考勤系統(tǒng)

• 權(quán)游播完了, 你在罵爛尾, 有人卻悄悄解鎖了新操作……
  

真香，朕在看了！

總結(jié)

以上是生活随笔為你收集整理的360深度实践：Flink与Storm协议级对比的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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