1. 為什么流計(jì)算繞不開(kāi)狀態(tài)這個(gè)坎很多人剛接觸流處理的時(shí)候腦子里想的都是數(shù)據(jù)一條條來(lái)我一條條處理完就完事了。這個(gè)思路在無(wú)狀態(tài)場(chǎng)景下確實(shí)沒(méi)問(wèn)題——比如單純做字段清洗、格式轉(zhuǎn)換、簡(jiǎn)單過(guò)濾來(lái)一條處理一條處理完就扔不需要記住任何東西。但只要業(yè)務(wù)稍微復(fù)雜一點(diǎn)比如要統(tǒng)計(jì)過(guò)去五分鐘每個(gè)用戶(hù)的點(diǎn)擊次數(shù)、要判斷這筆訂單是不是重復(fù)提交、要檢測(cè)某個(gè)設(shè)備連續(xù)三次上報(bào)異常溫度你就必須讓程序記住之前發(fā)生過(guò)什么。這個(gè)記住就是狀態(tài)。我見(jiàn)過(guò)不少團(tuán)隊(duì)在流計(jì)算項(xiàng)目里栽跟頭不是因?yàn)樗阕訉?xiě)得不對(duì)而是因?yàn)閷?duì)狀態(tài)的理解停留在表面。他們知道要用狀態(tài)但不知道狀態(tài)存在哪、怎么存、出問(wèn)題了怎么恢復(fù)、狀態(tài)會(huì)不會(huì)無(wú)限膨脹。結(jié)果上線跑了一段時(shí)間要么是任務(wù)重啟后數(shù)據(jù)對(duì)不上要么是內(nèi)存被狀態(tài)撐爆要么是 checkpoint 一直失敗導(dǎo)致整個(gè)作業(yè)不穩(wěn)定。這些問(wèn)題的根子都在狀態(tài)管理上。Flink 之所以在流計(jì)算領(lǐng)域站得住腳很大程度就是因?yàn)樗褷顟B(tài)管理這件事做透了。它不像有些框架那樣把狀態(tài)當(dāng)成一個(gè)附加功能而是從架構(gòu)層面就把狀態(tài)、容錯(cuò)、一致性這幾件事綁在一起設(shè)計(jì)。你寫(xiě)代碼的時(shí)候可能只是調(diào)了一個(gè)ValueState或者M(jìn)apState但背后涉及的是狀態(tài)后端選型、checkpoint 機(jī)制、狀態(tài)過(guò)期策略、序列化方案這一整套東西。這篇內(nèi)容適合兩類(lèi)人看一類(lèi)是已經(jīng)能寫(xiě) Flink 作業(yè)、但對(duì)狀態(tài)管理還是一知半解的同學(xué)另一類(lèi)是在生產(chǎn)環(huán)境被狀態(tài)相關(guān)問(wèn)題折騰過(guò)、想系統(tǒng)梳理一遍的工程師。我會(huì)從狀態(tài)的基本分類(lèi)講起然后深入到狀態(tài)后端、checkpoint 配置、狀態(tài)過(guò)期、大狀態(tài)優(yōu)化這些實(shí)戰(zhàn)中真正會(huì)踩坑的地方。不會(huì)只講 API 怎么調(diào)而是把為什么這么設(shè)計(jì)生產(chǎn)環(huán)境該怎么選這些事說(shuō)清楚。2. Flink 里的狀態(tài)到底分幾種各自解決什么問(wèn)題2.1 Keyed State 和 Operator State 的本質(zhì)區(qū)別Flink 把狀態(tài)分成兩大類(lèi)Keyed State鍵控狀態(tài)和Operator State算子狀態(tài)。這個(gè)劃分不是隨便定的它直接決定了狀態(tài)怎么分布、怎么擴(kuò)容、怎么恢復(fù)。Keyed State 只能用在KeyedStream上也就是你調(diào)過(guò)keyBy()之后的流。它的特點(diǎn)是每個(gè) key 維護(hù)一份獨(dú)立的狀態(tài)狀態(tài)和 key 是綁定的。比如你按用戶(hù) ID 分組統(tǒng)計(jì)消費(fèi)金額那每個(gè)用戶(hù) ID 就對(duì)應(yīng)一份狀態(tài)。這種狀態(tài)的好處是天然支持水平擴(kuò)展——當(dāng)并行度從 2 提到 4 的時(shí)候Flink 只需要把原來(lái)每個(gè) key 的狀態(tài)重新分配到新的并行實(shí)例上就行因?yàn)?key 和狀態(tài)是一一對(duì)應(yīng)的。Operator State 則是綁定在算子實(shí)例上的每個(gè)并行子任務(wù)維護(hù)自己的一份狀態(tài)。它不依賴(lài) key典型場(chǎng)景是 Kafka Source 里記錄每個(gè)分區(qū)消費(fèi)到哪個(gè) offset 了。這種狀態(tài)在改變并行度的時(shí)候恢復(fù)策略和 Keyed State 完全不同需要算子自己實(shí)現(xiàn)分配邏輯。我剛開(kāi)始學(xué)的時(shí)候老是把這兩個(gè)搞混后來(lái)用一個(gè)類(lèi)比就記住了Keyed State 像是按人名存的檔案柜每個(gè)人一個(gè)抽屜Operator State 像是每個(gè)工位自己的筆記本工位換了筆記本怎么交接得另說(shuō)。2.2 五種 Keyed State 的選型邏輯Keyed State 具體又分五種選哪種直接關(guān)系到內(nèi)存占用和訪問(wèn)效率狀態(tài)類(lèi)型存儲(chǔ)結(jié)構(gòu)適用場(chǎng)景注意事項(xiàng)ValueState單值存最新一條記錄、計(jì)數(shù)器每次讀寫(xiě)整個(gè)值ListState列表存最近 N 條記錄遍歷成本隨長(zhǎng)度增長(zhǎng)MapState鍵值對(duì)存維度關(guān)聯(lián)、去重集合推薦替代 ListState 做去重ReducingState單值聚合函數(shù)累加、求最值只進(jìn)不出無(wú)法遍歷AggregatingState單值聚合器復(fù)雜聚合如平均值輸入輸出類(lèi)型可不同這里有個(gè)特別容易踩的坑做去重的時(shí)候很多人用 ListState然后每次來(lái)數(shù)據(jù)遍歷一遍判斷是否存在。數(shù)據(jù)量小的時(shí)候沒(méi)事一旦去重集合漲到幾十萬(wàn)每次遍歷就是災(zāi)難。正確做法是用 MapStatekey 存去重標(biāo)識(shí)value 存?zhèn)€占位符判斷存在性是 O(1) 的。這個(gè)優(yōu)化我在一個(gè)實(shí)時(shí)去重項(xiàng)目里做過(guò)把處理延遲從秒級(jí)降到了毫秒級(jí)。ReducingState 和 AggregatingState 的區(qū)別也值得說(shuō)一下。ReducingState 的輸入和輸出類(lèi)型必須一樣比如都是 Long適合做累加。AggregatingState 的輸入輸出可以不同比如輸入是訂單金額輸出是總額/筆數(shù)這樣的平均值結(jié)構(gòu)它通過(guò)一個(gè) Accumulator 中間態(tài)來(lái)實(shí)現(xiàn)。如果你要算平均值、方差這類(lèi)需要多字段維護(hù)的聚合就得用 AggregatingState。2.3 狀態(tài)描述符的注冊(cè)時(shí)機(jī)定義狀態(tài)必須用StateDescriptor而且必須在open()方法里或者作為成員變量初始化。我見(jiàn)過(guò)有人在processElement里臨時(shí)創(chuàng)建 StateDescriptor這是大忌——每次處理數(shù)據(jù)都創(chuàng)建描述符不僅浪費(fèi)資源還可能導(dǎo)致?tīng)顟B(tài)無(wú)法正確恢復(fù)。public class CountWithState extends KeyedProcessFunctionString, Event, Long { // 成員變量方式聲明推薦 private ValueStateLong countState; Override public void open(Configuration parameters) { ValueStateDescriptorLong descriptor new ValueStateDescriptor(count, Long.class); countState getRuntimeContext().getState(descriptor); } Override public void processElement(Event event, Context ctx, CollectorLong out) throws Exception { Long current countState.value(); if (current null) { current 0L; } current 1; countState.update(current); out.collect(current); } }狀態(tài)名字上面代碼里的 count在同一個(gè)算子內(nèi)必須唯一而且一旦上線就不能隨便改因?yàn)?checkpoint 里存的狀態(tài)是按名字索引的。改名字等于讓 Flink 找不到原來(lái)的狀態(tài)恢復(fù)的時(shí)候會(huì)當(dāng)成全新?tīng)顟B(tài)處理歷史數(shù)據(jù)就丟了。這個(gè)坑我在早期項(xiàng)目里踩過(guò)一次改了個(gè)狀態(tài)名結(jié)果重啟后所有累計(jì)值歸零排查了半天才反應(yīng)過(guò)來(lái)。3. 狀態(tài)后端選型內(nèi)存、RocksDB 到底怎么選3.1 三種狀態(tài)后端的定位差異狀態(tài)后端決定了狀態(tài)存在哪、怎么存。Flink 主要有三種HashMapStateBackend把狀態(tài)存在 TaskManager 的 JVM 堆內(nèi)存里讀寫(xiě)就是操作 Java 對(duì)象速度極快。但缺點(diǎn)是狀態(tài)大小受限于堆內(nèi)存而且 GC 壓力大。狀態(tài)量在幾百 MB 以?xún)?nèi)、對(duì)延遲極其敏感的場(chǎng)景適合用它。EmbeddedRocksDBStateBackend把狀態(tài)存在 RocksDB 這個(gè)嵌入式 KV 數(shù)據(jù)庫(kù)里數(shù)據(jù)落在磁盤(pán)上內(nèi)存只做緩存。它能支撐 TB 級(jí)別的狀態(tài)但每次讀寫(xiě)涉及序列化和磁盤(pán) IO延遲比堆內(nèi)高。大狀態(tài)、對(duì)延遲不那么苛刻的場(chǎng)景選它。HashMapStateBackend 異步快照是現(xiàn)在比較推薦的組合兼顧了堆內(nèi)讀寫(xiě)速度和 checkpoint 時(shí)的穩(wěn)定性。選型的時(shí)候我會(huì)問(wèn)自己三個(gè)問(wèn)題狀態(tài)總量大概多大能接受多大的處理延遲checkpoint 頻率多高如果狀態(tài)超過(guò) 1GB我基本直接上 RocksDB不猶豫。堆內(nèi)狀態(tài)一旦超過(guò)堆的 50%GC 就會(huì)開(kāi)始影響吞吐而且 checkpoint 時(shí)同步快照會(huì)阻塞數(shù)據(jù)處理造成明顯的毛刺。3.2 RocksDB 的增量 checkpoint 為什么重要RocksDB 狀態(tài)后端最大的價(jià)值之一是支持增量 checkpoint。全量 checkpoint 每次都要把整個(gè)狀態(tài)寫(xiě)一遍狀態(tài)大了以后 checkpoint 時(shí)間可能比 checkpoint 間隔還長(zhǎng)導(dǎo)致作業(yè)一直在做 checkpoint沒(méi)法正常處理數(shù)據(jù)。增量 checkpoint 只上傳上次 checkpoint 之后變化的部分能把 checkpoint 時(shí)間降低一個(gè)數(shù)量級(jí)。配置增量 checkpoint 很簡(jiǎn)單EmbeddedRocksDBStateBackend backend new EmbeddedRocksDBStateBackend(true); // true 開(kāi)啟增量 env.setStateBackend(backend);但這里有個(gè)前提增量 checkpoint 需要配合支持持久化的存儲(chǔ)比如分布式文件系統(tǒng)。而且要注意增量 checkpoint 的元數(shù)據(jù)會(huì)隨著 checkpoint 數(shù)量增長(zhǎng)而膨脹需要定期做一次全量 checkpoint 來(lái)壓實(shí)。我一般會(huì)設(shè)置每 10 次增量后做一次全量通過(guò)state.backend.rocksdb.checkpoint.transfer.thread.num和相關(guān)的清理參數(shù)來(lái)控制。3.3 狀態(tài)后端的生產(chǎn)配置清單光選對(duì)后端還不夠參數(shù)調(diào)優(yōu)才是見(jiàn)功力的地方。下面這張表是我在多個(gè)項(xiàng)目里總結(jié)出來(lái)的關(guān)鍵配置配置項(xiàng)作用推薦值說(shuō)明state.backend狀態(tài)后端類(lèi)型rocksdb大狀態(tài)必選state.backend.incremental增量 checkpointtrueRocksDB 專(zhuān)屬state.backend.rocksdb.memory.managed托管內(nèi)存true避免 OOMstate.backend.rocksdb.block.cache-size塊緩存64-256MB根據(jù)內(nèi)存調(diào)state.checkpoints.dircheckpoint 路徑分布式存儲(chǔ)必須可靠state.checkpoints.num-retained保留數(shù)量3-5太多占空間state.backend.rocksdb.memory.managed這個(gè)參數(shù)特別值得說(shuō)。開(kāi)啟托管內(nèi)存后RocksDB 的內(nèi)存使用由 Flink 統(tǒng)一管理不會(huì)因?yàn)?RocksDB 自己亂申請(qǐng)內(nèi)存導(dǎo)致 TaskManager 被系統(tǒng)殺掉。我有個(gè)項(xiàng)目沒(méi)開(kāi)這個(gè)參數(shù)跑了一周后 TaskManager 頻繁重啟日志里全是內(nèi)存相關(guān)的錯(cuò)誤開(kāi)了托管內(nèi)存后就穩(wěn)定了。4. Checkpoint 機(jī)制狀態(tài)容錯(cuò)的核心鏈路4.1 從 Barrier 對(duì)齊看一致性保證Checkpoint 的本質(zhì)是給整個(gè)作業(yè)的狀態(tài)拍一張全局一致的快照。Flink 用的是Chandy-Lamport 算法的變體核心是往數(shù)據(jù)流里插入一種叫 Barrier 的特殊標(biāo)記。JobManager 定期觸發(fā) checkpoint往 Source 注入 Barrier。Barrier 隨著數(shù)據(jù)往下游流每個(gè)算子收到所有輸入流的 Barrier 后就把自己的狀態(tài)快照存下來(lái)然后繼續(xù)往下游傳 Barrier。當(dāng)所有算子的快照都完成后一個(gè)完整的 checkpoint 就生成了。這里有個(gè)關(guān)鍵概念叫Barrier 對(duì)齊。當(dāng)一個(gè)算子有多個(gè)輸入流時(shí)它收到第一個(gè)流的 Barrier 后不能立刻處理得等所有流的 Barrier 都到齊。等待期間先到 Barrier 的那個(gè)流的數(shù)據(jù)會(huì)被緩存起來(lái)。這個(gè)機(jī)制保證了 Exactly-Once 語(yǔ)義但代價(jià)是對(duì)齊期間會(huì)有延遲。如果業(yè)務(wù)能接受 At-Least-Once可以開(kāi)啟非對(duì)齊 checkpointUnaligned Checkpoint。它不等 Barrier 對(duì)齊直接把緩存中的數(shù)據(jù)也一起快照進(jìn)去大幅降低反壓場(chǎng)景下的 checkpoint 延遲。配置方式env.getCheckpointConfig().enableUnalignedCheckpoints();非對(duì)齊 checkpoint 在反壓嚴(yán)重的時(shí)候效果特別明顯。我做過(guò)一個(gè)測(cè)試同樣的反壓場(chǎng)景下對(duì)齊 checkpoint 要 30 秒才能完成非對(duì)齊只要 3 秒。但它的代價(jià)是 checkpoint 體積會(huì)變大因?yàn)橐?in-flight 的數(shù)據(jù)也存進(jìn)去。4.2 Checkpoint 配置的取舍藝術(shù)Checkpoint 配置沒(méi)有標(biāo)準(zhǔn)答案得根據(jù)業(yè)務(wù)對(duì)延遲、吞吐、恢復(fù)時(shí)間的要求來(lái)權(quán)衡。下面是我常用的配置模板和背后的思考CheckpointConfig config env.getCheckpointConfig(); // 間隔太短影響吞吐太長(zhǎng)恢復(fù)慢 config.setCheckpointInterval(60_000); // 1分鐘 // 超時(shí)超過(guò)這個(gè)時(shí)間就放棄本次 config.setCheckpointTimeout(10 * 60_000); // 10分鐘 // 并發(fā)同一時(shí)刻只允許一個(gè) config.setMaxConcurrentCheckpoints(1); // 兩次之間最小間隔給作業(yè)喘息時(shí)間 config.setMinPauseBetweenCheckpoints(30_000); // 失敗容忍連續(xù)失敗幾次才讓作業(yè)失敗 config.setTolerableCheckpointFailureNumber(3); // 外部化作業(yè)取消后保留 checkpoint config.enableExternalizedCheckpoints( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);setMinPauseBetweenCheckpoints這個(gè)參數(shù)很多人忽略。如果不設(shè)checkpoint 一個(gè)接一個(gè)作業(yè)幾乎沒(méi)有時(shí)間處理正常數(shù)據(jù)吞吐會(huì)明顯下降。設(shè)一個(gè)合理的間隔讓作業(yè)在兩次 checkpoint 之間有穩(wěn)定的處理窗口整體表現(xiàn)會(huì)好很多。setTolerableCheckpointFailureNumber也很實(shí)用。生產(chǎn)環(huán)境偶爾會(huì)有網(wǎng)絡(luò)抖動(dòng)、存儲(chǔ)短暫不可用如果一次失敗就讓作業(yè)掛掉運(yùn)維壓力太大。容忍幾次失敗給系統(tǒng)自愈的機(jī)會(huì)但也不能設(shè)太大否則真出問(wèn)題了發(fā)現(xiàn)太晚。4.3 保存點(diǎn)與 checkpoint 的區(qū)別很多人把 Savepoint 和 Checkpoint 混為一談其實(shí)它們用途不同。Checkpoint 是 Flink 自動(dòng)管理的用于故障恢復(fù)作業(yè)停止后默認(rèn)會(huì)清理。Savepoint 是手動(dòng)觸發(fā)的用于作業(yè)升級(jí)、遷移、擴(kuò)縮容會(huì)一直保留直到你手動(dòng)刪除。做作業(yè)升級(jí)的時(shí)候正確流程是先觸發(fā) Savepoint停掉舊作業(yè)用新代碼從 Savepoint 恢復(fù)。這樣狀態(tài)能平滑遷移。但要注意Savepoint 恢復(fù)對(duì)算子的 UID 有要求。如果你沒(méi)給算子設(shè)置穩(wěn)定的 UIDFlink 會(huì)用自動(dòng)生成的代碼一改 UID 就變了Savepoint 就恢復(fù)不了。DataStreamEvent stream env.addSource(new KafkaSource()) .uid(kafka-source) // 手動(dòng)指定穩(wěn)定 UID .map(new EnrichFunction()) .uid(enrich-map) .keyBy(Event::getUserId) .process(new CountFunction()) .uid(count-process);給每個(gè)算子設(shè) UID 是我強(qiáng)烈建議的習(xí)慣尤其是會(huì)用到 Savepoint 的場(chǎng)景。這個(gè)習(xí)慣能省掉大量升級(jí)時(shí)的麻煩。5. 狀態(tài)過(guò)期與清理別讓狀態(tài)無(wú)限膨脹5.1 TTL 機(jī)制的適用邊界流作業(yè)是 7x24 跑的狀態(tài)只增不減的話遲早會(huì)把存儲(chǔ)撐爆。Flink 提供了State TTL機(jī)制可以給狀態(tài)設(shè)置過(guò)期時(shí)間過(guò)期的狀態(tài)在讀取時(shí)會(huì)被清理。StateTtlConfig ttlConfig StateTtlConfig .newBuilder(Time.hours(24)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .cleanupInRocksdbCompactFilter(1000) .build(); ValueStateDescriptorLong descriptor new ValueStateDescriptor(count, Long.class); descriptor.enableTimeToLive(ttlConfig);TTL 的更新類(lèi)型有兩種OnCreateAndWrite表示創(chuàng)建和寫(xiě)入時(shí)刷新過(guò)期時(shí)間OnReadAndWrite表示讀取也刷新。選哪個(gè)取決于業(yè)務(wù)語(yǔ)義。如果是用戶(hù)最近一次活躍后 24 小時(shí)過(guò)期用OnReadAndWrite如果是狀態(tài)創(chuàng)建后固定 24 小時(shí)過(guò)期用OnCreateAndWrite。可見(jiàn)性也有講究。NeverReturnExpired表示過(guò)期狀態(tài)絕不返回語(yǔ)義干凈但清理有延遲ReturnExpiredIfNotCleanedUp表示如果還沒(méi)被物理清理過(guò)期狀態(tài)也能讀到性能好但語(yǔ)義模糊。我一般用NeverReturnExpired避免業(yè)務(wù)邏輯被過(guò)期數(shù)據(jù)干擾。5.2 TTL 不是萬(wàn)能的幾個(gè)失效場(chǎng)景TTL 看著美好但有幾個(gè)坑必須知道。第一TTL 的清理是惰性的。Flink 不會(huì)主動(dòng)掃描所有狀態(tài)去刪過(guò)期的而是在讀取時(shí)判斷。這意味著如果某個(gè) key 再也不被訪問(wèn)它的狀態(tài)會(huì)一直占著空間。對(duì)于 key 空間很大但訪問(wèn)稀疏的場(chǎng)景TTL 基本沒(méi)用。第二RocksDB 的 TTL 清理依賴(lài) compaction。只有配置了cleanupInRocksdbCompactFilter過(guò)期狀態(tài)才會(huì)在 RocksDB 做 compaction 時(shí)被真正刪除。如果不配過(guò)期狀態(tài)只是邏輯上不可見(jiàn)物理上還在。第三ListState 和 MapState 的 TTL 是元素級(jí)的。ListState 里每個(gè)元素獨(dú)立過(guò)期MapState 里每個(gè) entry 獨(dú)立過(guò)期。這有時(shí)候不符合預(yù)期——你可能希望整個(gè)列表一起過(guò)期但 TTL 做不到。對(duì)于 TTL 搞不定的場(chǎng)景我通常用定時(shí)器 手動(dòng)清理。比如用KeyedProcessFunction注冊(cè)一個(gè)定時(shí)器到時(shí)間了主動(dòng)clear()狀態(tài)。這種方式控制精確但要注意定時(shí)器本身也占狀態(tài)得管理好。5.3 大狀態(tài)作業(yè)的監(jiān)控指標(biāo)狀態(tài)管理做得好不好得靠監(jiān)控說(shuō)話。下面這些指標(biāo)我每次上線都會(huì)盯著指標(biāo)含義健康范圍checkpoint 大小單次快照體積穩(wěn)定或緩慢增長(zhǎng)checkpoint 耗時(shí)完成一次的時(shí)間小于間隔的 50%checkpoint 失敗率失敗次數(shù)占比接近 0狀態(tài)訪問(wèn)延遲讀寫(xiě)狀態(tài)耗時(shí)毫秒級(jí)RocksDB 后臺(tái)錯(cuò)誤compaction 異常0checkpoint 大小持續(xù)快速增長(zhǎng)通常意味著狀態(tài)泄漏得檢查 TTL 或者清理邏輯。checkpoint 耗時(shí)接近甚至超過(guò)間隔說(shuō)明狀態(tài)太大或者存儲(chǔ)太慢要考慮增量 checkpoint 或者換存儲(chǔ)。這些指標(biāo)在 Flink Web UI 和 Metrics 系統(tǒng)里都能看到建議接入告警。6. 生產(chǎn)環(huán)境狀態(tài)問(wèn)題的排查實(shí)錄6.1 一次 checkpoint 持續(xù)失敗的完整排查有個(gè)項(xiàng)目上線后checkpoint 一直失敗作業(yè)雖然還在跑但沒(méi)有任何容錯(cuò)能力隨時(shí)可能丟數(shù)據(jù)。我按下面的鏈路排查了一遍。第一步看失敗原因。Web UI 上顯示是 Checkpoint expired before completing意思是 checkpoint 超時(shí)了。這說(shuō)明 checkpoint 做不完不是做錯(cuò)了。第二步看 checkpoint 大小和耗時(shí)。發(fā)現(xiàn) checkpoint 大小有 20 多 GB耗時(shí)接近 15 分鐘而間隔設(shè)的是 10 分鐘。問(wèn)題清楚了狀態(tài)太大checkpoint 根本做不完。第三步分析狀態(tài)為什么這么大。查代碼發(fā)現(xiàn)有個(gè)去重邏輯用了 ListState而且沒(méi)有 TTL。這個(gè)去重集合隨著時(shí)間無(wú)限增長(zhǎng)跑了一周就攢了上億條記錄。第四步定位根因。ListState 做去重本身就是錯(cuò)誤選型應(yīng)該用 MapState。而且沒(méi)有 TTL歷史數(shù)據(jù)永遠(yuǎn)不清理。第五步修復(fù)。把 ListState 換成 MapState加上 7 天 TTL同時(shí)開(kāi)啟 RocksDB 增量 checkpoint。改完后 checkpoint 大小降到 2GB耗時(shí) 1 分鐘以?xún)?nèi)問(wèn)題解決。這個(gè)案例的教訓(xùn)是狀態(tài)選型和 TTL 必須在設(shè)計(jì)階段就想清楚不能等出問(wèn)題了再補(bǔ)。ListState 做去重、無(wú) TTL 的大集合這兩個(gè)都是典型的狀態(tài)反模式。6.2 狀態(tài)恢復(fù)后數(shù)據(jù)對(duì)不上的詭異問(wèn)題另一個(gè)項(xiàng)目遇到的現(xiàn)象是作業(yè)從 checkpoint 恢復(fù)后統(tǒng)計(jì)結(jié)果和恢復(fù)前對(duì)不上有些 key 的計(jì)數(shù)明顯偏小。排查過(guò)程比較繞。首先確認(rèn) checkpoint 是成功的恢復(fù)日志也沒(méi)報(bào)錯(cuò)。然后對(duì)比恢復(fù)前后的狀態(tài)發(fā)現(xiàn)確實(shí)少了一部分。后來(lái)發(fā)現(xiàn)問(wèn)題出在狀態(tài)名字沖突上。這個(gè)作業(yè)里有兩個(gè)不同的算子都用了默認(rèn)的狀態(tài)名而 Flink 在某些情況下會(huì)把它們混淆。雖然概率不高但一旦發(fā)生就會(huì)導(dǎo)致?tīng)顟B(tài)錯(cuò)亂。解決辦法是給每個(gè)狀態(tài)起全局唯一且語(yǔ)義清晰的名字比如user-click-count、order-dedup-set這種而不是state1、count這種容易撞車(chē)的名字。同時(shí)給算子設(shè) UID雙保險(xiǎn)。這個(gè)坑很隱蔽因?yàn)榇蟛糠謺r(shí)候不會(huì)出問(wèn)題只有在特定條件下才觸發(fā)。但一旦觸發(fā)排查起來(lái)非常痛苦。所以我現(xiàn)在養(yǎng)成了習(xí)慣狀態(tài)名和算子 UID 都認(rèn)真起絕不用默認(rèn)值。6.3 反壓導(dǎo)致 checkpoint 雪崩的應(yīng)對(duì)反壓和 checkpoint 是一對(duì)冤家。反壓嚴(yán)重時(shí)Barrier 在流里流動(dòng)得很慢checkpoint 遲遲完不成超時(shí)失敗。失敗的 checkpoint 又會(huì)占用資源進(jìn)一步加劇反壓形成惡性循環(huán)。應(yīng)對(duì)這個(gè)問(wèn)題我總結(jié)了幾招第一招開(kāi)非對(duì)齊 checkpoint。前面說(shuō)過(guò)它不等 Barrier 對(duì)齊能大幅緩解反壓下的 checkpoint 延遲。這是最直接有效的手段。第二招調(diào)大 checkpoint 超時(shí)和間隔。給 checkpoint 更多時(shí)間別讓它頻繁失敗。但這只是緩解治標(biāo)不治本。第三招從根上解決反壓。反壓的本質(zhì)是下游處理不過(guò)來(lái)要么加并行度要么優(yōu)化算子邏輯要么做數(shù)據(jù)傾斜處理。checkpoint 問(wèn)題往往是反壓的表象得治本。第四招監(jiān)控反壓指標(biāo)。Flink 的backPressure指標(biāo)能告訴你哪個(gè)算子反壓了順著往下查就能找到瓶頸。我一般會(huì)先開(kāi)非對(duì)齊 checkpoint 穩(wěn)住局面然后花時(shí)間定位反壓根因。只治 checkpoint 不治反壓遲早還會(huì)出問(wèn)題。7. 狀態(tài)編程的幾條實(shí)戰(zhàn)心得聊了這么多機(jī)制和案例最后分享幾條我在實(shí)際項(xiàng)目里沉淀下來(lái)的經(jīng)驗(yàn)都是踩過(guò)坑才明白的。狀態(tài)能小則小。設(shè)計(jì)狀態(tài)結(jié)構(gòu)的時(shí)候只存真正需要的信息。比如存用戶(hù) ID 就夠了別把整個(gè)用戶(hù)對(duì)象塞進(jìn)去。狀態(tài)里每多一個(gè)字段序列化開(kāi)銷(xiāo)、存儲(chǔ)開(kāi)銷(xiāo)、checkpoint 開(kāi)銷(xiāo)都會(huì)增加。我見(jiàn)過(guò)有人把整個(gè)訂單對(duì)象存狀態(tài)里其實(shí)只需要訂單 ID 和金額兩個(gè)字段。序列化器要選對(duì)。Flink 對(duì)狀態(tài)序列化有要求用錯(cuò)序列化器會(huì)導(dǎo)致性能問(wèn)題甚至兼容性問(wèn)題?；绢?lèi)型用 Flink 自帶的序列化器就行復(fù)雜對(duì)象建議用 Avro 或者自己實(shí)現(xiàn)TypeSerializer。千萬(wàn)別用 Java 原生序列化又慢又不兼容。狀態(tài)訪問(wèn)要節(jié)制。每次value()、update()都是有成本的尤其是在 RocksDB 后端下涉及磁盤(pán) IO。能在內(nèi)存里緩存的中間結(jié)果就別反復(fù)讀寫(xiě)狀態(tài)。我有個(gè)優(yōu)化案例把循環(huán)里的狀態(tài)讀寫(xiě)提到循環(huán)外性能提升了 3 倍。升級(jí)前先測(cè) Savepoint。任何涉及狀態(tài)結(jié)構(gòu)變更的升級(jí)都要先在測(cè)試環(huán)境驗(yàn)證 Savepoint 能不能正常恢復(fù)。狀態(tài)結(jié)構(gòu)改了、狀態(tài)名改了、算子 UID 改了都可能導(dǎo)致恢復(fù)失敗。提前測(cè)別等上線了才發(fā)現(xiàn)。監(jiān)控要覆蓋狀態(tài)維度。除了常規(guī)的吞吐、延遲指標(biāo)一定要監(jiān)控 checkpoint 大小、耗時(shí)、失敗率以及狀態(tài)訪問(wèn)的相關(guān)指標(biāo)。這些指標(biāo)能提前預(yù)警狀態(tài)問(wèn)題讓你在作業(yè)崩潰前介入處理。狀態(tài)編程這件事入門(mén)容易精通難。API 就那么幾個(gè)但要用好得理解背后的機(jī)制得知道生產(chǎn)環(huán)境會(huì)遇到什么。希望這些內(nèi)容能幫你少走點(diǎn)彎路。真到了生產(chǎn)環(huán)境遇到具體問(wèn)題還是要結(jié)合監(jiān)控和日志具體分析沒(méi)有銀彈。