1. 從一個(gè)反直覺的現(xiàn)象說起為什么我的Flink SQL任務(wù)重啟后數(shù)據(jù)對(duì)不上剛接觸Flink SQL那會(huì)兒我踩過一個(gè)印象特別深的坑。當(dāng)時(shí)寫了一個(gè)很簡(jiǎn)單的聚合任務(wù)從Kafka讀訂單流按用戶ID做累加結(jié)果寫到MySQL。本地測(cè)試一切正常上線跑了三天某次機(jī)房網(wǎng)絡(luò)抖動(dòng)導(dǎo)致任務(wù)重啟恢復(fù)之后發(fā)現(xiàn)MySQL里的累計(jì)金額比實(shí)際業(yè)務(wù)數(shù)據(jù)多了一大截。排查了大半天才反應(yīng)過來——問題出在我對(duì)動(dòng)態(tài)表這個(gè)概念的理解上。我當(dāng)時(shí)腦子里還是傳統(tǒng)數(shù)據(jù)庫(kù)那套思維表就是磁盤上的一堆數(shù)據(jù)查詢就是對(duì)這些數(shù)據(jù)做計(jì)算。但Flink SQL里的表根本不是這么回事它是隨著數(shù)據(jù)流不斷變化的動(dòng)態(tài)表查詢結(jié)果也是動(dòng)態(tài)的而結(jié)果表到底怎么輸出、輸出多少次取決于你選的查詢類型和輸出模式。這個(gè)認(rèn)知偏差直接導(dǎo)致了重復(fù)累加。后來我把動(dòng)態(tài)表的連續(xù)查詢語(yǔ)義、回撤流機(jī)制、時(shí)間屬性這幾塊徹底捋了一遍才算真正把Flink SQL用明白了。這篇文章就把我這些年對(duì)動(dòng)態(tài)表的理解完整梳理一遍從概念本質(zhì)到實(shí)操細(xì)節(jié)再到那些文檔里不會(huì)明說的坑盡量講透。如果你正在用Flink SQL做實(shí)時(shí)計(jì)算或者準(zhǔn)備從DataStream API遷移到SQL這篇內(nèi)容應(yīng)該能幫你少走不少?gòu)澛贰?. 動(dòng)態(tài)表到底動(dòng)態(tài)在哪里流與表的對(duì)偶關(guān)系拆解2.1 傳統(tǒng)表與動(dòng)態(tài)表的本質(zhì)差異先把這個(gè)概念說清楚不然后面全是空中樓閣。傳統(tǒng)數(shù)據(jù)庫(kù)里的表你可以理解成一張靜態(tài)快照——在某個(gè)時(shí)刻執(zhí)行SELECT * FROM orders返回的是那一刻表里的全部數(shù)據(jù)查詢完就結(jié)束了結(jié)果集是固定的。但Flink SQL面對(duì)的是源源不斷的流數(shù)據(jù)數(shù)據(jù)不是已經(jīng)在那里的而是正在到來的。Flink的做法是把流看成表的變更日志changelog。一條INSERT對(duì)應(yīng)流里的一條新增記錄一條UPDATE對(duì)應(yīng)一條更新通常拆成一條DELETE加一條INSERT一條DELETE對(duì)應(yīng)一條刪除記錄。這樣一張表在任意時(shí)刻的內(nèi)容就是從開始到現(xiàn)在所有變更累積后的結(jié)果。隨著新數(shù)據(jù)不斷到來這張表的內(nèi)容持續(xù)變化所以叫動(dòng)態(tài)表。這里有個(gè)關(guān)鍵點(diǎn)很多人會(huì)忽略動(dòng)態(tài)表本身不存儲(chǔ)數(shù)據(jù)它是對(duì)流的一種邏輯抽象。你寫的SELECT語(yǔ)句作用在動(dòng)態(tài)表上產(chǎn)生的也是一個(gè)動(dòng)態(tài)表Flink會(huì)把對(duì)這個(gè)結(jié)果動(dòng)態(tài)表的查詢持續(xù)翻譯成對(duì)流數(shù)據(jù)的連續(xù)計(jì)算。這就是所謂的連續(xù)查詢Continuous Query。2.2 連續(xù)查詢?yōu)槭裁从啦煌Ｖ箓鹘y(tǒng)查詢執(zhí)行一次就返回結(jié)果連續(xù)查詢不一樣它一旦啟動(dòng)就永遠(yuǎn)在跑因?yàn)檩斎肓鳑]有結(jié)束的概念除非你顯式設(shè)置了有界流。每來一條新數(shù)據(jù)連續(xù)查詢就可能更新一次結(jié)果。這就引出一個(gè)核心問題結(jié)果怎么更新Flink用**回撤流Retract Stream和追加流Append Stream**兩種模式來描述結(jié)果的更新方式。追加流好理解就是只增不改每條結(jié)果都是新的?；爻妨鲝?fù)雜一些它用兩種消息表示更新一條Iadd表示新增一條-Uretract表示撤回舊值然后可能再跟一條U表示新值。你如果用過Flink的toRetractStream看到的就是這種帶布爾標(biāo)記的二元組。理解這一點(diǎn)至關(guān)重要因?yàn)樗苯記Q定了你的下游系統(tǒng)能不能正確處理結(jié)果。比如你把回撤流寫到一個(gè)只支持追加的Kafka topic里那些撤回消息就會(huì)變成普通數(shù)據(jù)混進(jìn)去下游一讀就亂套了。2.3 一個(gè)具體例子把抽象概念落地光說理論太干舉個(gè)實(shí)際場(chǎng)景。假設(shè)有一張訂單動(dòng)態(tài)表orders字段是user_id和amount。我寫一個(gè)查詢SELECT user_id, SUM(amount) AS total FROM orders GROUP BY user_id這個(gè)查詢產(chǎn)生的結(jié)果動(dòng)態(tài)表每當(dāng)orders里來一條新訂單對(duì)應(yīng)user_id的total就會(huì)變化。假設(shè)用戶A先來了100元結(jié)果表里是(A, 100)又來了一筆50元結(jié)果表里(A, 100)這條要撤回變成(A, 150)。在回撤流里你看到的就是I (A, 100) -U (A, 100) U (A, 150)而如果查詢是SELECT user_id, amount FROM orders這種不帶聚合的簡(jiǎn)單投影結(jié)果就是純追加流每條輸入對(duì)應(yīng)一條輸出沒有撤回。這個(gè)區(qū)別看起來簡(jiǎn)單但它決定了你整個(gè)任務(wù)的輸出語(yǔ)義。我見過太多人在這里翻車尤其是做實(shí)時(shí)大屏或者寫數(shù)據(jù)庫(kù)的時(shí)候沒搞清楚結(jié)果流類型導(dǎo)致數(shù)據(jù)重復(fù)或者丟失。3. 時(shí)間屬性動(dòng)態(tài)表能正確計(jì)算的前提3.1 處理時(shí)間與事件時(shí)間的取舍邏輯動(dòng)態(tài)表要處理的是無界流而無界流里數(shù)據(jù)到達(dá)的順序和實(shí)際發(fā)生的順序往往不一致。這就引出了時(shí)間屬性的概念。Flink SQL里主要有兩種時(shí)間處理時(shí)間Processing Time和事件時(shí)間Event Time。處理時(shí)間就是數(shù)據(jù)被Flink處理時(shí)的機(jī)器時(shí)間簡(jiǎn)單、無需額外配置但結(jié)果不確定——同樣的數(shù)據(jù)不同時(shí)間跑出來的結(jié)果可能不一樣。事件時(shí)間則是數(shù)據(jù)里自帶的時(shí)間戳反映事件真實(shí)發(fā)生的時(shí)間能保證結(jié)果的一致性但需要處理亂序和延遲數(shù)據(jù)。我的經(jīng)驗(yàn)是對(duì)結(jié)果準(zhǔn)確性要求高的場(chǎng)景比如計(jì)費(fèi)、對(duì)賬、風(fēng)控必須用事件時(shí)間對(duì)實(shí)時(shí)性要求高、能容忍一定誤差的場(chǎng)景比如實(shí)時(shí)監(jiān)控大盤處理時(shí)間夠用且更簡(jiǎn)單。這個(gè)取舍沒有標(biāo)準(zhǔn)答案取決于業(yè)務(wù)對(duì)正確的定義。3.2 水位線事件時(shí)間能推進(jìn)的引擎用事件時(shí)間就必須理解水位線Watermark。水位線是一個(gè)時(shí)間戳表示我認(rèn)為所有時(shí)間戳小于等于這個(gè)值的數(shù)據(jù)都已經(jīng)到齊了。它像是一個(gè)進(jìn)度條推動(dòng)著事件時(shí)間往前走。為什么需要水位線因?yàn)榫酆喜僮鞅热绨创翱谇蠛捅仨氈朗裁磿r(shí)候可以輸出結(jié)果。如果永遠(yuǎn)等下去那結(jié)果永遠(yuǎn)出不來。水位線就是告訴Flink這個(gè)時(shí)間點(diǎn)之前的數(shù)據(jù)我基本收齊了可以觸發(fā)計(jì)算了。水位線的生成策略很關(guān)鍵。你可以用WATERMARK FOR ts AS ts - INTERVAL 5 SECOND這種固定延遲也可以用自定義的生成邏輯。延遲設(shè)太小遲到數(shù)據(jù)會(huì)被丟棄設(shè)太大結(jié)果輸出延遲高。我一般會(huì)先統(tǒng)計(jì)業(yè)務(wù)數(shù)據(jù)的實(shí)際亂序程度取P99的延遲作為水位線延遲再留一點(diǎn)余量。3.3 時(shí)間屬性在DDL里怎么聲明在Flink SQL的DDL里聲明時(shí)間屬性有固定寫法。事件時(shí)間字段必須配合水位線一起定義CREATE TABLE orders ( user_id BIGINT, amount DECIMAL(10, 2), order_time TIMESTAMP(3), WATERMARK FOR order_time AS order_time - INTERVAL 5 SECOND ) WITH ( connector kafka, ... );處理時(shí)間則不需要水位線直接用PROCTIME()函數(shù)生成一個(gè)虛擬列CREATE TABLE orders ( user_id BIGINT, amount DECIMAL(10, 2), proc_time AS PROCTIME() ) WITH (...);這里有個(gè)容易踩的坑時(shí)間屬性字段的類型必須是TIMESTAMP(3)或TIMESTAMP_LTZ(3)精度是毫秒。如果你從Kafka讀進(jìn)來的時(shí)間戳是BIGINT類型的毫秒數(shù)得先用TO_TIMESTAMP_LTZ之類的函數(shù)轉(zhuǎn)換不能直接當(dāng)時(shí)間屬性用。我當(dāng)初就因?yàn)樽侄晤愋筒粚?duì)水位線死活不生效排查了好久。4. 動(dòng)態(tài)表上的三類查詢能做什么不能做什么4.1 簡(jiǎn)單投影與過濾最安全的追加流SELECT加WHERE這種查詢是最簡(jiǎn)單的結(jié)果動(dòng)態(tài)表只會(huì)有新增不會(huì)有更新或刪除。因?yàn)槊織l輸入記錄獨(dú)立產(chǎn)生一條輸出記錄不涉及跨記錄的狀態(tài)。這類查詢產(chǎn)生的是追加流下游處理起來最省心。但要注意即使是簡(jiǎn)單投影如果涉及計(jì)算字段也要考慮計(jì)算邏輯的冪等性。比如amount * 1.1這種沒問題但如果用了RAND()這種非確定性函數(shù)結(jié)果就不可復(fù)現(xiàn)了。4.2 聚合查詢回撤流的重災(zāi)區(qū)GROUP BY加聚合函數(shù)是動(dòng)態(tài)表最核心也最容易出問題的部分。前面說過聚合結(jié)果會(huì)隨著新數(shù)據(jù)不斷更新產(chǎn)生回撤流。這里有幾個(gè)關(guān)鍵限制第一不是所有聚合都支持更新。像SUM、COUNT、AVG、MAX、MIN這些支持增量更新的沒問題。但有些聚合函數(shù)在流模式下不支持或者需要特殊配置。第二GROUP BY的鍵決定了狀態(tài)大小。如果按用戶ID分組狀態(tài)里要維護(hù)每個(gè)用戶的累計(jì)值。用戶量上億的話狀態(tài)會(huì)非常大必須配置好狀態(tài)后端和TTL。我見過一個(gè)任務(wù)因?yàn)闆]設(shè)狀態(tài)TTL跑了半個(gè)月后狀態(tài)膨脹到幾百GB直接把TaskManager撐爆了。第三普通聚合和窗口聚合的輸出語(yǔ)義不同。普通GROUP BY是持續(xù)更新的每條新數(shù)據(jù)都可能觸發(fā)結(jié)果更新窗口聚合則是窗口結(jié)束時(shí)輸出一次。這個(gè)區(qū)別在寫下游的時(shí)候必須搞清楚。4.3 雙流JOIN動(dòng)態(tài)表最復(fù)雜的場(chǎng)景雙流JOIN是動(dòng)態(tài)表里最燒腦的部分。兩條無界流做JOIN理論上任何一條流的新數(shù)據(jù)都可能和另一條流的歷史數(shù)據(jù)匹配這意味著兩邊都得把歷史數(shù)據(jù)存下來狀態(tài)無限增長(zhǎng)。Flink對(duì)雙流JOIN做了限制主要支持區(qū)間JOINInterval Join和帶時(shí)間條件的JOIN。比如SELECT o.user_id, o.amount, p.product_name FROM orders o JOIN products p ON o.product_id p.product_id AND o.order_time BETWEEN p.update_time - INTERVAL 1 HOUR AND p.update_time INTERVAL 1 HOUR這個(gè)時(shí)間區(qū)間條件就是用來限制狀態(tài)范圍的——超出區(qū)間的數(shù)據(jù)可以被清理掉。如果你不加時(shí)間條件直接JOINFlink會(huì)報(bào)錯(cuò)或者要求你配置狀態(tài)TTL否則狀態(tài)會(huì)無限增長(zhǎng)。我的實(shí)操建議是雙流JOIN一定要帶時(shí)間區(qū)間條件而且區(qū)間要盡可能小。區(qū)間越大需要保留的狀態(tài)越多內(nèi)存壓力越大。同時(shí)要監(jiān)控狀態(tài)大小設(shè)置合理的TTL。5. 動(dòng)態(tài)表到外部系統(tǒng)的輸出那些文檔沒講透的細(xì)節(jié)5.1 不同查詢類型對(duì)應(yīng)的輸出模式把動(dòng)態(tài)表的結(jié)果寫到外部系統(tǒng)時(shí)輸出模式必須和查詢類型匹配。這是個(gè)硬性約束搞錯(cuò)了要么報(bào)錯(cuò)要么數(shù)據(jù)錯(cuò)亂。查詢類型結(jié)果流類型支持的輸出模式典型下游簡(jiǎn)單投影/過濾追加流AppendKafka、文件系統(tǒng)聚合查詢回撤流Upsert / Retract數(shù)據(jù)庫(kù)、KV存儲(chǔ)窗口聚合追加流窗口結(jié)束時(shí)AppendKafka、OLAP雙流JOIN追加流或回撤流視情況視情況Upsert模式要求下游支持按主鍵更新比如MySQL、HBase、Redis。Retract模式則把撤回消息也當(dāng)成普通數(shù)據(jù)發(fā)出去需要下游自己能處理。Append模式最簡(jiǎn)單只發(fā)新增。我一般優(yōu)先選Upsert模式寫數(shù)據(jù)庫(kù)因?yàn)檎Z(yǔ)義最清晰下游不用關(guān)心撤回邏輯。但前提是你的查詢結(jié)果有明確的主鍵而且下游支持upsert。5.2 主鍵聲明的坑在DDL里聲明主鍵不只是為了語(yǔ)義清晰它直接影響輸出行為。比如寫MySQL時(shí)CREATE TABLE user_total ( user_id BIGINT, total DECIMAL(10, 2), PRIMARY KEY (user_id) NOT ENFORCED ) WITH ( connector jdbc, ... );NOT ENFORCED表示Flink不強(qiáng)制校驗(yàn)主鍵唯一性但會(huì)用這個(gè)主鍵信息來決定upsert行為。如果你不聲明主鍵Flink就不知道按什么鍵更新可能退化成append模式導(dǎo)致數(shù)據(jù)重復(fù)。這里有個(gè)細(xì)節(jié)主鍵聲明必須和查詢的GROUP BY鍵一致否則upsert會(huì)出問題。比如你按user_id分組結(jié)果表主鍵也得是user_id。5.3 冪等性寫入的保障即使配置都對(duì)了任務(wù)重啟時(shí)還是可能產(chǎn)生重復(fù)數(shù)據(jù)。因?yàn)镕link的checkpoint機(jī)制保證的是至少一次at-least-once語(yǔ)義要做到精確一次exactly-once需要下游支持事務(wù)或者冪等寫入。對(duì)于數(shù)據(jù)庫(kù)寫入我通常的做法是用upsert模式并且保證主鍵唯一。這樣即使同一條數(shù)據(jù)寫兩次結(jié)果也是一樣的天然冪等。如果下游不支持upsert那就得在業(yè)務(wù)層面做去重比如加一個(gè)唯一索引或者用版本號(hào)。Kafka作為下游時(shí)精確一次需要開啟事務(wù)配置sink.semantic exactly-once但這會(huì)增加延遲。我的經(jīng)驗(yàn)是除非業(yè)務(wù)強(qiáng)要求否則at-least-once加冪等寫入就夠了。6. 狀態(tài)管理動(dòng)態(tài)表背后的隱形戰(zhàn)場(chǎng)6.1 狀態(tài)為什么會(huì)膨脹動(dòng)態(tài)表的連續(xù)查詢需要維護(hù)中間狀態(tài)。聚合查詢要存每個(gè)分組的累計(jì)值雙流JOIN要存兩邊流的歷史數(shù)據(jù)去重查詢要存已經(jīng)見過的key。這些狀態(tài)默認(rèn)存在內(nèi)存里數(shù)據(jù)量一大就出問題。狀態(tài)膨脹的根源是無界流沒有終點(diǎn)。傳統(tǒng)批處理跑完就釋放內(nèi)存流處理得一直扛著。如果業(yè)務(wù)數(shù)據(jù)的key空間很大比如按用戶ID分組用戶上億狀態(tài)就會(huì)持續(xù)增長(zhǎng)。6.2 狀態(tài)TTL的配置策略Flink SQL支持通過table.exec.state.ttl配置狀態(tài)過期時(shí)間。比如設(shè)成1 HOUR表示狀態(tài)里的數(shù)據(jù)超過1小時(shí)沒更新就被清理掉。-- 在Flink配置里設(shè)置 table.exec.state.ttl: 3600000但TTL不是設(shè)了就好得考慮業(yè)務(wù)語(yǔ)義。比如你算的是用戶歷史累計(jì)消費(fèi)設(shè)了TTL之后超過1小時(shí)沒消費(fèi)的用戶之前的累計(jì)值就被清了下次再來消費(fèi)就從零開始算結(jié)果就錯(cuò)了。所以TTL只適用于那些只看近期數(shù)據(jù)的場(chǎng)景比如最近1小時(shí)的訂單量。我的建議是先想清楚業(yè)務(wù)對(duì)時(shí)間范圍的要求再?zèng)Q定TTL。如果業(yè)務(wù)需要全量累計(jì)那就得接受大狀態(tài)配置好RocksDB狀態(tài)后端和足夠的磁盤。如果業(yè)務(wù)只看近期那就大膽設(shè)TTL能省很多資源。6.3 RocksDB狀態(tài)后端的調(diào)優(yōu)經(jīng)驗(yàn)狀態(tài)大到內(nèi)存放不下時(shí)就得用RocksDB狀態(tài)后端把狀態(tài)存到磁盤。但RocksDB的默認(rèn)配置不一定適合所有場(chǎng)景我踩過幾個(gè)坑第一state.backend.rocksdb.block.cache-size要調(diào)大。默認(rèn)值偏小讀性能上不去。我一般設(shè)成可用內(nèi)存的1/4到1/3。第二state.backend.rocksdb.writebuffer.size和writebuffer.count要配合調(diào)。寫buffer太小會(huì)導(dǎo)致頻繁flush影響吞吐。第三定期做compaction。RocksDB的LSM樹結(jié)構(gòu)會(huì)導(dǎo)致讀放大后臺(tái)compaction跟不上時(shí)讀性能會(huì)急劇下降。可以通過state.backend.rocksdb.compaction.style選擇適合的壓縮策略。這些參數(shù)沒有萬能值得根據(jù)實(shí)際數(shù)據(jù)量和訪問模式調(diào)。我一般會(huì)先用默認(rèn)配置跑觀察checkpoint大小和延遲再針對(duì)性調(diào)整。7. 幾個(gè)真實(shí)踩過的坑和排查思路7.1 結(jié)果重復(fù)寫入從checkpoint機(jī)制找原因前面提到的重復(fù)累加問題根因是任務(wù)重啟后從checkpoint恢復(fù)但checkpoint之后到故障點(diǎn)之間的數(shù)據(jù)被重新處理了。如果下游不是冪等的就會(huì)重復(fù)。排查思路先確認(rèn)下游是否支持upsert主鍵是否聲明正確。如果都對(duì)了還重復(fù)檢查checkpoint間隔是不是太長(zhǎng)——間隔越長(zhǎng)故障時(shí)重放的數(shù)據(jù)越多?？梢赃m當(dāng)縮短checkpoint間隔但也不能太短否則影響吞吐。7.2 水位線不推進(jìn)數(shù)據(jù)類型是元兇有次任務(wù)跑著跑著結(jié)果不輸出了排查發(fā)現(xiàn)水位線卡住不動(dòng)。最后定位到時(shí)間字段類型不對(duì)——從Kafka讀進(jìn)來的是字符串雖然內(nèi)容看著像時(shí)間但Flink不認(rèn)水位線生成器直接失效。解決辦法是在DDL里用計(jì)算列做類型轉(zhuǎn)換CREATE TABLE orders ( order_time_str STRING, order_time AS TO_TIMESTAMP(order_time_str), WATERMARK FOR order_time AS order_time - INTERVAL 5 SECOND ) WITH (...);這個(gè)坑的教訓(xùn)是時(shí)間屬性字段的類型必須嚴(yán)格匹配不能想當(dāng)然。7.3 狀態(tài)太大導(dǎo)致任務(wù)頻繁重啟TTL和key設(shè)計(jì)有個(gè)任務(wù)按(user_id, product_id)分組key空間是用戶數(shù)乘商品數(shù)直接爆炸。后來改成先按user_id聚合再按product_id聚合分兩步走狀態(tài)小了很多。這個(gè)案例說明GROUP BY的key設(shè)計(jì)直接影響狀態(tài)大小。能分步聚合的就別一步到位能減少key維度的就盡量減少。8. 我對(duì)動(dòng)態(tài)表這套機(jī)制的理解和日常使用建議用久了會(huì)發(fā)現(xiàn)動(dòng)態(tài)表這套抽象其實(shí)是在用表的思維來簡(jiǎn)化流的編程。你不用再手動(dòng)管理狀態(tài)、處理亂序、寫窗口邏輯SQL層幫你把這些都封裝了。但封裝不等于消失底層的狀態(tài)、時(shí)間、回撤這些機(jī)制還在只是換了個(gè)形式暴露給你。我的日常使用習(xí)慣是寫SQL之前先想清楚三件事——數(shù)據(jù)有沒有時(shí)間屬性、查詢會(huì)不會(huì)產(chǎn)生回撤、狀態(tài)會(huì)不會(huì)無限增長(zhǎng)。這三個(gè)問題想明白了大部分坑都能提前避開。另外別迷信SQL的簡(jiǎn)潔。有些復(fù)雜邏輯用DataStream API寫反而更清晰、更可控。SQL適合那些模式固定、邏輯相對(duì)簡(jiǎn)單的場(chǎng)景比如ETL、簡(jiǎn)單聚合、維表關(guān)聯(lián)。真正復(fù)雜的業(yè)務(wù)邏輯該用API還是得用API。最后分享一個(gè)調(diào)試技巧用EXPLAIN看執(zhí)行計(jì)劃。Flink SQL的EXPLAIN能告訴你查詢被翻譯成了什么樣的算子拓?fù)溆袥]有產(chǎn)生回撤、狀態(tài)用在哪里、有沒有不必要的shuffle。我每次寫完一個(gè)稍復(fù)雜的查詢都會(huì)先EXPLAIN一下確認(rèn)執(zhí)行計(jì)劃符合預(yù)期再上線。這個(gè)習(xí)慣幫我省了很多事后排查的時(shí)間。