聊流計(jì)算之前我先說(shuō)一個(gè)我真實(shí)經(jīng)歷過(guò)的場(chǎng)景。某次業(yè)務(wù)方提了一個(gè)需求做一個(gè)實(shí)時(shí)大屏要求數(shù)據(jù)從產(chǎn)生到出現(xiàn)在大屏上不超過(guò)5秒鐘。我下意識(shí)打開(kāi)之前的離線(xiàn)調(diào)度任務(wù)準(zhǔn)備沿用每天凌晨跑一次、第二天看結(jié)果的方式結(jié)果被業(yè)務(wù)方一句話(huà)頂了回來(lái)我要的是現(xiàn)在不是昨天。就是從那次開(kāi)始我才真正意識(shí)到流計(jì)算解決的核心矛盾不是算得快而是數(shù)據(jù)從產(chǎn)生到被消費(fèi)之間的時(shí)間差。批處理把這些差值壓縮到T1流計(jì)算則追求數(shù)據(jù)到達(dá)即處理結(jié)果以秒級(jí)甚至毫秒級(jí)延遲對(duì)外可見(jiàn)。在流計(jì)算這個(gè)領(lǐng)域里繞不開(kāi)四個(gè)名字Flink、Storm、Spark Streaming還有一個(gè)Spark。很多剛接觸的人會(huì)把它們當(dāng)成并列關(guān)系實(shí)際上Spark本身是批處理引擎Spark Streaming只是它上面的流處理擴(kuò)展而Flink和Storm從一開(kāi)始就是為實(shí)時(shí)數(shù)據(jù)設(shè)計(jì)的。這篇文章不打算做教科書(shū)式的原理宣講我想從實(shí)際選型和使用者的視角把這四個(gè)名字背后的處理模型、實(shí)戰(zhàn)路徑和真實(shí)代價(jià)講清楚盡量讓準(zhǔn)備入手流計(jì)算的人少走彎路。1. 從明天看報(bào)表到現(xiàn)在看數(shù)據(jù)流計(jì)算解決的核心矛盾流式計(jì)算的價(jià)值本質(zhì)上取決于業(yè)務(wù)對(duì)數(shù)據(jù)新鮮度的要求。今天很多場(chǎng)景已經(jīng)不是數(shù)據(jù)有沒(méi)有的問(wèn)題而是數(shù)據(jù)晚到幾秒鐘業(yè)務(wù)就受不了的問(wèn)題。1.1 批處理為什么在很多場(chǎng)景下不夠用批處理的核心思路是攢夠一批再算。離線(xiàn)任務(wù)把一天的數(shù)據(jù)通過(guò)調(diào)度系統(tǒng)在某個(gè)時(shí)間點(diǎn)集中拉取經(jīng)過(guò)清洗、關(guān)聯(lián)、聚合最終產(chǎn)出報(bào)表。它的問(wèn)題不是不能算而是算得太晚。舉個(gè)例子一個(gè)交易系統(tǒng)想識(shí)別同一張銀行卡在短時(shí)間內(nèi)高頻消費(fèi)的異常行為如果等T1的批處理跑完用戶(hù)在第二天才收到風(fēng)控提示這個(gè)提示已經(jīng)失去了攔截意義。另外一個(gè)容易忽略的點(diǎn)是批處理雖然延遲高但它對(duì)資源的使用往往是峰值集中的。每天凌晨跑批時(shí)計(jì)算資源被打滿(mǎn)白天則大量空閑。這種資源模型遇到實(shí)時(shí)性要求高的業(yè)務(wù)時(shí)完全失效因?yàn)閿?shù)據(jù)是持續(xù)產(chǎn)生的處理系統(tǒng)也需要持續(xù)運(yùn)行而不是集中在某個(gè)時(shí)間節(jié)點(diǎn)集中爆發(fā)計(jì)算。1.2 流計(jì)算最典型的幾類(lèi)應(yīng)用場(chǎng)景流計(jì)算最常見(jiàn)的落地場(chǎng)景我總結(jié)下來(lái)有三類(lèi)。第一類(lèi)是實(shí)時(shí)監(jiān)控和可視化比如大屏上的交易量、在線(xiàn)用戶(hù)數(shù)、設(shè)備狀態(tài)這類(lèi)場(chǎng)景對(duì)延遲極其敏感5秒以上就會(huì)覺(jué)得不實(shí)時(shí)。第二類(lèi)是實(shí)時(shí)風(fēng)控和反欺詐系統(tǒng)需要根據(jù)用戶(hù)行為序列判斷風(fēng)險(xiǎn)數(shù)據(jù)每多延遲一秒風(fēng)險(xiǎn)擴(kuò)散的范圍就大一分。第三類(lèi)是實(shí)時(shí)ETL和指標(biāo)計(jì)算把業(yè)務(wù)數(shù)據(jù)庫(kù)的變更通過(guò)變更數(shù)據(jù)捕獲同步到數(shù)據(jù)倉(cāng)庫(kù)或下游系統(tǒng)這時(shí)候流的角色是數(shù)據(jù)管道而不是分析引擎。還有一個(gè)場(chǎng)景是流式推薦。用戶(hù)點(diǎn)擊、瀏覽、加購(gòu)的行為不斷產(chǎn)生推薦系統(tǒng)需要根據(jù)最近幾分鐘的行為實(shí)時(shí)調(diào)整推送內(nèi)容。這就是為什么很多互聯(lián)網(wǎng)公司對(duì)流計(jì)算的需求非常旺盛——他們的業(yè)務(wù)本身就是現(xiàn)在發(fā)生什么現(xiàn)在就要響應(yīng)什么。1.3 實(shí)時(shí)這個(gè)詞在不同框架里的含義不同這里要先埋一個(gè)伏筆不同流處理框架對(duì)實(shí)時(shí)的定義完全不一樣。Storm是純事件驅(qū)動(dòng)的每條數(shù)據(jù)到達(dá)后立刻被處理延遲在毫秒級(jí)Spark Streaming把流切成秒級(jí)的小批次所以它的延遲下限就是一個(gè)批次的大小Flink則介于兩者之間它是真正的流式處理但也支持按時(shí)間窗口做聚合延遲取決于窗口策略。后面我會(huì)詳細(xì)展開(kāi)這三個(gè)模型的區(qū)別現(xiàn)在只需要建立一個(gè)大概念實(shí)時(shí)不是一種狀態(tài)而是一個(gè)范圍選型之前先要搞清楚業(yè)務(wù)到底能容忍多少秒的延遲。2. 三大引擎的本質(zhì)差異流式、微批次、純事件驅(qū)動(dòng)分別意味著什么把Flink、Storm、Spark Streaming放在一張表里對(duì)比是理解它們差異最快的方式。我先把結(jié)論擺出來(lái)然后逐個(gè)拆解。維度StormSpark StreamingFlink處理模型原生流式逐條處理微批次Micro-batch原生流式批流一體延遲毫秒級(jí)秒級(jí)取決于批次間隔毫秒到秒級(jí)取決于窗口狀態(tài)管理較弱多依賴(lài)外部存儲(chǔ)支持基于StateStore內(nèi)置狀態(tài)后端支持大狀態(tài)時(shí)間語(yǔ)義處理時(shí)間為主處理時(shí)間為主后支持事件時(shí)間事件時(shí)間加水位線(xiàn)原生支持精確一次語(yǔ)義需結(jié)合外部系統(tǒng)實(shí)現(xiàn)Spark 2.x起支持內(nèi)置checkpoint原生支持API抽象Spout/Bolt偏底層DStream / Structured StreamingDataStream / Table SQL2.1 Storm真正的一條一條處理Storm是我最早接觸的流式框架。它的核心抽象是拓?fù)湟粋€(gè)拓?fù)溆蒘pout和Bolt組成Spout從數(shù)據(jù)源讀取消息Bolt負(fù)責(zé)處理每個(gè)Spout或Bolt在集群上并行運(yùn)行多個(gè)實(shí)例。Storm的原始模型就是數(shù)據(jù)一條一條流過(guò)不攢批、不等待天然適合毫秒級(jí)延遲的場(chǎng)景。但這種設(shè)計(jì)的代價(jià)是為了提高吞吐Storm對(duì)消息的可靠性保障做得比較吃力。它通過(guò)一種ack機(jī)制來(lái)追蹤每條消息是否被完整處理如果超時(shí)未確認(rèn)就重新發(fā)送。這個(gè)機(jī)制在低流量下沒(méi)問(wèn)題一旦流量暴漲大量消息元的追蹤本身就會(huì)成為性能瓶頸。Storm后來(lái)推出的Trident抽象就是為了解決可靠性問(wèn)題但它實(shí)際上又回到了微批次模型等于把優(yōu)勢(shì)讓了出去。2.2 Spark Streaming把流切成批來(lái)處理Spark Streaming的思路完全是另一種角度。它不做逐條處理而是把連續(xù)不斷的流按照設(shè)定的間隔比如1秒或5秒切成一段一段的小批次每一批對(duì)應(yīng)Spark的一個(gè)作業(yè)再交給Spark Core去算。這個(gè)過(guò)程對(duì)用戶(hù)是透明的你寫(xiě)的是DStream操作底層跑的卻是批處理作業(yè)。微批次模型的優(yōu)點(diǎn)非常明顯它繼承了Spark Core的容錯(cuò)機(jī)制、調(diào)度機(jī)制和豐富的算子開(kāi)發(fā)門(mén)檻比Storm低很多。而且由于批次是定長(zhǎng)的系統(tǒng)的吞吐可以做得比較高資源利用也更穩(wěn)定。缺點(diǎn)同樣明顯延遲被批次間隔鎖死就算你把批次設(shè)成1秒也意味著必須攢夠1秒的數(shù)據(jù)才能觸發(fā)一次計(jì)算做不到逐條響應(yīng)的毫秒級(jí)延遲。2.3 Flink流式計(jì)算優(yōu)先批處理是流式的一種特例Flink的設(shè)計(jì)哲學(xué)跟Storm和Spark Streaming都不一樣。Flink認(rèn)為流式是數(shù)據(jù)的天然形態(tài)批處理只是流的一個(gè)邊界情況——把所有數(shù)據(jù)當(dāng)作一個(gè)有限流來(lái)處理。所以在Flink里核心的數(shù)據(jù)模型是無(wú)限流加上窗口Window窗口把無(wú)界數(shù)據(jù)切分成有界范圍來(lái)做聚合和關(guān)聯(lián)。Flink對(duì)事件時(shí)間Event Time的支持是它相比前兩者最大的技術(shù)護(hù)城河。真實(shí)世界中數(shù)據(jù)產(chǎn)生的時(shí)間和處理的時(shí)間經(jīng)常不一致消息可能在網(wǎng)絡(luò)里亂序延遲。Flink用事件時(shí)間和水位線(xiàn)Watermark機(jī)制處理亂序問(wèn)題在窗口計(jì)算中等待遲到的數(shù)據(jù)同時(shí)保證實(shí)時(shí)性。這套機(jī)制在金融、風(fēng)控等對(duì)時(shí)序敏感的領(lǐng)域幾乎是剛需而Storm和早期的Spark Streaming都做不到這么精細(xì)。3. Flink實(shí)戰(zhàn)本地環(huán)境、核心API與checkpoint配置我不打算空談原理直接分享一段我曾經(jīng)跑通的Flink流式作業(yè)。如果你身邊有一臺(tái)電腦哪怕配置一般也可以跟著走一遍。3.1 本地環(huán)境準(zhǔn)備不需要先搭集群很多人一聽(tīng)到Flink就以為必須部署一套集群才能跑這是一個(gè)很大的誤區(qū)。Flink支持本地模式最簡(jiǎn)單的做法是下載發(fā)行包然后啟動(dòng)一個(gè)本地集群或者干脆在IDE里用Maven/Gradle建一個(gè)工程依賴(lài)?yán)镆隖link的Java客戶(hù)端庫(kù)即可。我推薦第二種方式因?yàn)榉奖阏{(diào)試代碼跟生產(chǎn)環(huán)境是同一套API。需要的依賴(lài)大致是flink-streaming-java和flink-clients版本建議統(tǒng)一。我在本地用Gradle建工程時(shí)grouptoplevel只引入兩個(gè)模塊一個(gè)是org.apache.flink:flink-streaming-java另一個(gè)是org.apache.flink:flink-clients運(yùn)行環(huán)境是JDK8就能把最簡(jiǎn)單的流作業(yè)跑起來(lái)。這里有一個(gè)容易踩的坑Flink對(duì)Scala和Java的版本有對(duì)應(yīng)要求如果你用Scala API一定要注意Scala二進(jìn)制版本是否匹配否則運(yùn)行時(shí)會(huì)報(bào)方法不存在之類(lèi)的異常。3.2 第一個(gè)實(shí)時(shí)WordCount用Socket當(dāng)數(shù)據(jù)源下面這個(gè)例子是流式WordCount數(shù)據(jù)源是Socket端口。打開(kāi)一個(gè)終端執(zhí)行nc -lk 9999隨便輸入一些單詞Flink作業(yè)就會(huì)實(shí)時(shí)累計(jì)每個(gè)單詞出現(xiàn)的次數(shù)。先看代碼import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; public class StreamWordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment .getExecutionEnvironment(); DataStreamString text env.socketTextStream(localhost, 9999); DataStreamTuple2String, Integer counts text .flatMap((String line, CollectorTuple2String, Integer out) - { for (String word : line.split(\\s)) { out.collect(new Tuple2(word, 1)); } }) .returns(Types.TUPLE(Types.STRING, Types.INT)) .keyBy(tuple - tuple.f0) .sum(1); counts.print(); env.execute(Stream WordCount); } }這段代碼里有幾個(gè)關(guān)鍵信息。socketTextStream表示數(shù)據(jù)源是不間斷的流keyBy把相同單詞分到同一個(gè)分區(qū)sum(1)對(duì)Tuple2的第二個(gè)字段做累加。注意.returns(...)不是可有可無(wú)的Java的Lambda表達(dá)式在Flink里經(jīng)常因?yàn)轭?lèi)型擦除丟信息如果不顯式指定返回類(lèi)型作業(yè)提交時(shí)可能直接報(bào)類(lèi)型錯(cuò)誤。這里有一個(gè)新手容易忽略的點(diǎn)print()在本地模式下直接輸出到控制臺(tái)但在集群上它會(huì)輸出到TaskManager的日志里。很多人在集群上跑完發(fā)現(xiàn)控制臺(tái)沒(méi)輸出東西以為任務(wù)沒(méi)執(zhí)行其實(shí)數(shù)據(jù)早就寫(xiě)到日志中了。調(diào)試時(shí)用print沒(méi)問(wèn)題生產(chǎn)環(huán)境建議直接接到Kafka或者別的下游系統(tǒng)。3.3 checkpoint與狀態(tài)后端上生產(chǎn)前必須做的事本地跑通WordCount只是第一步。真正把Flink作業(yè)部署到生產(chǎn)環(huán)境前有兩件事必須做配置checkpoint和選好狀態(tài)后端。checkpoint是Flink容錯(cuò)的基石它周期性地把任務(wù)的狀態(tài)和位置保存到持久存儲(chǔ)中一旦任務(wù)失敗就從最近一次checkpoint恢復(fù)。沒(méi)有checkpoint的Flink作業(yè)一旦節(jié)點(diǎn)宕機(jī)狀態(tài)會(huì)丟得一干二凈這對(duì)于有狀態(tài)計(jì)算是致命的。配置方式很簡(jiǎn)單在創(chuàng)建環(huán)境之后加幾行env.enableCheckpointing(10000); // 每10秒做一次checkpoint env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000); env.getCheckpointConfig().setCheckpointTimeout(60000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); env.setStateBackend(new EmbeddedRocksDBStateBackend());這里我解釋一下幾個(gè)參數(shù)的意義。enableCheckpointing(10000)是checkpoint周期周期越短恢復(fù)時(shí)丟失的數(shù)據(jù)越少但checkpoint本身也會(huì)占用資源setMinPauseBetweenCheckpoints是兩次checkpoint之間至少間隔多久防止前一個(gè)還沒(méi)完成就啟動(dòng)下一個(gè)setCheckpointTimeout是單次checkpoint的超市時(shí)間防止?fàn)顟B(tài)太大導(dǎo)致deadline無(wú)限拉長(zhǎng)。狀態(tài)后端我推薦RocksDB它能支持超大狀態(tài)把狀態(tài)落到本地磁盤(pán)而不是全部擠在內(nèi)存里。還有一個(gè)跟事件時(shí)間相關(guān)的坑。Flink要處理亂序事件需要設(shè)置水位線(xiàn)策略。常見(jiàn)做法是WatermarkStrategyString strategy WatermarkStrategy .StringforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, timestamp) - extractTimestamp(event));forBoundedOutOfOrderness(5秒)的意思是我允許最多5秒的亂序超過(guò)5秒才算遲到。這個(gè)值設(shè)得越大窗口觸發(fā)越晚實(shí)時(shí)性越差設(shè)得越小遲到的數(shù)據(jù)越多計(jì)算結(jié)果越不準(zhǔn)。具體設(shè)多少要根據(jù)數(shù)據(jù)源的亂序程度來(lái)測(cè)不能拍腦袋。4. Spark Streaming的兩條路線(xiàn)DStream和Structured Streaming怎么選Spark生態(tài)里跟流計(jì)算相關(guān)的有兩套API老牌的DStream和后來(lái)的Structured Streaming。很多人一上來(lái)就被這兩個(gè)名字搞暈我先說(shuō)結(jié)論新項(xiàng)目盡量用Structured StreamingDStream已經(jīng)處于維護(hù)狀態(tài)只在老項(xiàng)目里會(huì)遇到。4.1 DStream API經(jīng)典RDD序列DStream的全稱(chēng)是Discretized Stream離散化流。它的核心思想是把連續(xù)流按時(shí)間間隔切分每一個(gè)間隔內(nèi)的數(shù)據(jù)形成一個(gè)RDD然后對(duì)RDD做各種轉(zhuǎn)換操作。你可以把DStream理解為RDD的序列每一批數(shù)據(jù)本質(zhì)上是Spark Core在處理。寫(xiě)DStream代碼的體驗(yàn)跟寫(xiě)批處理很接近。比如同樣是WordCountDStream版本大致是這樣val stream ssc.receiverStream(new SocketReceiver(localhost, 9999)) stream .flatMap(_.split( )) .map(word (word, 1)) .reduceByKey(_ _)問(wèn)題在于DStream的窗口操作、狀態(tài)操作全部基于處理時(shí)間Processing Time它感知不到事件產(chǎn)生的時(shí)間。如果你的業(yè)務(wù)數(shù)據(jù)存在延遲到達(dá)的情況DStream處理出來(lái)的結(jié)果會(huì)漂移得很厲害。我在一個(gè)日志分析項(xiàng)目里就遇到過(guò)日志落盤(pán)時(shí)間跟Kafka收到時(shí)間差了十幾分鐘按處理時(shí)間做窗口結(jié)果完全對(duì)不上業(yè)務(wù)口徑。4.2 Structured Streaming從流到表的抽象Structured Streaming把流式數(shù)據(jù)當(dāng)成一張不斷追加的表來(lái)處理。每一秒到達(dá)的新數(shù)據(jù)就是往表里插入新行而用戶(hù)的查詢(xún)語(yǔ)句寫(xiě)在這張表上引擎會(huì)持續(xù)輸出結(jié)果。這個(gè)抽象最大的好處是你可以用處理DataFrame的思維來(lái)處理流API是統(tǒng)一的。同樣的WordCount用Structured Streaming的Python API寫(xiě)出來(lái)是這樣from pyspark.sql import SparkSession spark SparkSession.builder.appName(word_count).getOrCreate() lines spark.readStream.format(socket) \ .option(host, localhost) \ .option(port, 9999) \ .load() words lines.selectExpr(explode(split(value, )) as word) counts words.groupBy(word).count() query counts.writeStream \ .outputMode(complete) \ .format(console) \ .start() query.awaitTermination()注意readStream和writeStream這兩個(gè)方法它們對(duì)應(yīng)批處理里的read和write。Structured Streaming支持事件時(shí)間窗口、水位線(xiàn)還支持用append、update、complete三種輸出模式控制結(jié)果表的寫(xiě)法。相比DStream它在語(yǔ)義上完整得多。4.3 Spark Streaming的適用邊界在哪里雖然Structured Streaming是當(dāng)前推薦路線(xiàn)但我不認(rèn)為老的Spark Streaming就一無(wú)是處。如果你已有的Spark批處理任務(wù)非常多團(tuán)隊(duì)對(duì)DataFrame API相當(dāng)熟想用同一套棧覆蓋批和流那么Spark生態(tài)天然是首選。它的優(yōu)勢(shì)在于構(gòu)建在Spark Core之上歷史包袱少、社區(qū)用戶(hù)多、排查問(wèn)題資料豐富。代價(jià)就是延遲下不來(lái)特別在意毫秒級(jí)響應(yīng)的場(chǎng)景不適合它。我個(gè)人的經(jīng)驗(yàn)是如果延遲可以接受2秒以上且整體技術(shù)棧已經(jīng)重度綁定了Spark那就沒(méi)必要硬上Flink如果業(yè)務(wù)對(duì)延遲和精確性要求都很高那就不要糾結(jié)直接選Flink。5. Storm的價(jià)值重估低延遲之外還有哪些不可替代性很多人現(xiàn)在聊流計(jì)算基本都聚焦在Flink和Spark Streaming上Storm好像成了上古技術(shù)。我覺(jué)得這種判斷有點(diǎn)一刀切。Storm確實(shí)在主流實(shí)時(shí)計(jì)算場(chǎng)景中逐漸退場(chǎng)但它代表的純事件驅(qū)動(dòng)理念和它身上的一些特性在某些場(chǎng)景下仍然有用。5.1 Spout和Bolt的拓?fù)浣Y(jié)構(gòu)Storm的程序模型是拓?fù)?。拓?fù)淅镒詈诵牡膬蓚€(gè)組件是Spout和BoltSpout是數(shù)據(jù)源負(fù)責(zé)從Kafka、消息隊(duì)列或日志文件中讀取消息Bolt是處理節(jié)點(diǎn)可以執(zhí)行過(guò)濾、聚合、關(guān)聯(lián)等操作。多個(gè)Spout和Bolt通過(guò)數(shù)據(jù)流連接成一張有向圖就是整個(gè)拓?fù)?。這種模型非常直觀(guān)。我曾在某個(gè)物聯(lián)網(wǎng)項(xiàng)目里用Storm處理設(shè)備上報(bào)的傳感數(shù)據(jù)每條數(shù)據(jù)經(jīng)過(guò)解析、清洗、閾值判斷三個(gè)Bolt就能觸發(fā)告警。因?yàn)椴粩€批數(shù)據(jù)到達(dá)后確實(shí)能做到毫秒級(jí)處理這在當(dāng)時(shí)的物理網(wǎng)設(shè)備告警場(chǎng)景下是剛需。5.2 ack機(jī)制如何保證可靠性Storm為了保證消息不丟設(shè)計(jì)了一套acker機(jī)制。每個(gè)Spout發(fā)出的消息帶上一個(gè)消息IDacker負(fù)責(zé)追蹤這條消息經(jīng)過(guò)的所有Bolt是否執(zhí)行成功。如果某條消息在超時(shí)時(shí)間內(nèi)得不到應(yīng)答Spout會(huì)重新發(fā)送。這里有一個(gè)典型權(quán)衡開(kāi)啟ack機(jī)制會(huì)影響吞吐因?yàn)槊恳粭l消息都需要在acker層面記錄狀態(tài)。如果關(guān)閉ack吞吐大幅提升但消息丟失概率也上升。我在一個(gè)流量超大的場(chǎng)景中測(cè)試過(guò)開(kāi)啟ack后吞吐下降將近一半。所以Storm的使用者必須做好取舍不是每條消息都需要可靠保障的。5.3 哪些場(chǎng)景還在用Storm目前還在堅(jiān)持用Storm的場(chǎng)景我觀(guān)察下來(lái)主要是兩類(lèi)。第一類(lèi)是已經(jīng)把大量拓?fù)渑芊€(wěn)定了、遷移成本極高的存量系統(tǒng)它們沒(méi)有足夠的動(dòng)力重寫(xiě)。第二類(lèi)是對(duì)延遲極度敏感且消息量不是特別大的場(chǎng)景比如證券行情推送、實(shí)時(shí)告警。這類(lèi)場(chǎng)景里Storm的毫秒級(jí)處理能力和簡(jiǎn)潔模型反而比Flink的厚重更合適。不過(guò)客觀(guān)地說(shuō)Flink在延遲和可靠性?xún)煞矫娑家呀?jīng)超越了Storm且運(yùn)維生態(tài)更完善。如果你是一個(gè)新項(xiàng)目我不太建議再?gòu)腟torm起步。了解Storm的價(jià)值更多是為了讀老代碼或者在技術(shù)方案答辯時(shí)能說(shuō)清楚為什么不用Storm。6. 選型不是背參數(shù)結(jié)合業(yè)務(wù)場(chǎng)景的最終權(quán)衡與踩坑記錄技術(shù)選型的本質(zhì)是把業(yè)務(wù)需求翻譯成引擎能力。不同引擎的參數(shù)差異網(wǎng)上到處都有但怎么把這些差異映射到具體業(yè)務(wù)才是真正考驗(yàn)人的地方。6.1 延遲優(yōu)先場(chǎng)景風(fēng)控和交易系統(tǒng)我做過(guò)一個(gè)交易風(fēng)控項(xiàng)目要求從交易發(fā)生到風(fēng)險(xiǎn)判定必須在1秒內(nèi)完成。這個(gè)需求直接排除了Spark Streaming因?yàn)榫退惆雅伍g隔壓到500毫秒加上計(jì)算和調(diào)度開(kāi)銷(xiāo)也很難穩(wěn)定控制在1秒內(nèi)。當(dāng)時(shí)我在Flink和Storm之間猶豫。最終選Flink的原因有兩個(gè)一是我們需要根據(jù)用戶(hù)的跨渠道行為做關(guān)聯(lián)分析這需要強(qiáng)大的狀態(tài)管理能力Flink內(nèi)置狀態(tài)后端顯然比Storm的亂糟糟實(shí)現(xiàn)更可靠二是Flink的exactly-once語(yǔ)義能避免重復(fù)計(jì)算Storm在極端情況下可能重復(fù)觸發(fā)風(fēng)控規(guī)則這對(duì)用戶(hù)是極大的困擾。6.2 吞吐優(yōu)先場(chǎng)景日志ETL和數(shù)據(jù)管道如果業(yè)務(wù)只是把埋點(diǎn)日志從Kafka實(shí)時(shí)搬運(yùn)到數(shù)據(jù)倉(cāng)庫(kù)中間做一些解析、過(guò)濾、字段映射對(duì)延遲要求是別超過(guò)一兩分鐘那Flink和Spark Streaming都能干。這種場(chǎng)景我傾向于選Spark生態(tài)因?yàn)閿?shù)據(jù)倉(cāng)庫(kù)本身就是Spark的天下復(fù)用同一套資源池和調(diào)度體系可以降低運(yùn)維復(fù)雜度。尤其是如果下游還連著Spark SQL做即席查詢(xún)統(tǒng)一為Spark技術(shù)棧會(huì)省很多溝通成本。6.3 團(tuán)隊(duì)技術(shù)棧與運(yùn)維成本選型必須考慮團(tuán)隊(duì)里的人是寫(xiě)Java多還是Scala多、之前有沒(méi)有維護(hù)過(guò)實(shí)時(shí)集群的經(jīng)驗(yàn)。Flink的Java API非常成熟運(yùn)維上需要注意的狀態(tài)后端、checkpoint、反壓監(jiān)控等概念比較多Spark Streaming則依賴(lài)Spark集群運(yùn)維體系跟離線(xiàn)任務(wù)是一致的。如果團(tuán)隊(duì)里只有一個(gè)懂實(shí)時(shí)計(jì)算的人我建議優(yōu)先選擇這個(gè)團(tuán)隊(duì)最熟的技術(shù)棧而不是根據(jù)基準(zhǔn)測(cè)試報(bào)告里那幾毫秒的差距來(lái)決策。實(shí)時(shí)系統(tǒng)的故障排查遠(yuǎn)比性能峰值更消耗人力。6.4 我在幾個(gè)項(xiàng)目里踩過(guò)的實(shí)際坑最后分享幾個(gè)真實(shí)踩過(guò)的坑。第一個(gè)坑是Kafka消費(fèi)者的并行度設(shè)置。Flink里Kafka Source的并行度受限于Kafka Topic的分區(qū)數(shù)如果你把Source并行度設(shè)得比分區(qū)數(shù)還大多出來(lái)的并行度是空轉(zhuǎn)。反過(guò)來(lái)如果想提升吞吐先加Kafka分區(qū)再調(diào)Flink并行度。第二個(gè)坑是窗口計(jì)算的輸出延遲。很多人設(shè)了5分鐘的滾動(dòng)窗口以為第5分鐘一到就能出結(jié)果。實(shí)際上如果數(shù)據(jù)源存在亂序窗口要等水位線(xiàn)經(jīng)過(guò)才真正觸發(fā)。我在測(cè)試環(huán)境里用亂序數(shù)據(jù)驗(yàn)證過(guò)最終輸出比預(yù)期的窗口結(jié)束時(shí)間晚了將近半分鐘這是正常現(xiàn)象不是Bug。第三個(gè)坑是狀態(tài)后端的選擇。RocksDB能存大狀態(tài)但訪(fǎng)問(wèn)速度比堆內(nèi)存慢如果狀態(tài)小且內(nèi)存充裕用HashMap狀態(tài)后端的性能更好。我見(jiàn)過(guò)有項(xiàng)目把所有狀態(tài)都塞進(jìn)RocksDB結(jié)果吞吐上不去換回堆內(nèi)存瞬間就好了。實(shí)時(shí)計(jì)算是個(gè)跟業(yè)務(wù)強(qiáng)相關(guān)的領(lǐng)域每個(gè)場(chǎng)景的坑都不太一樣。這恰恰是它的有意思之處比的不只是能不能算還比在真實(shí)數(shù)據(jù)面前能不能穩(wěn)。上面這些經(jīng)驗(yàn)和代碼都是我一次次跑到生產(chǎn)環(huán)境后排出來(lái)的。如果你正在幾個(gè)引擎之間搖擺最好的辦法不是反復(fù)看對(duì)比文章而是把真實(shí)業(yè)務(wù)的一小段流量接到候選引擎上跑幾天用數(shù)據(jù)說(shuō)話(huà)。