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