戰(zhàn))
1. 時間語義選錯整條實(shí)時鏈路都在白跑實(shí)時計算里最容易被低估的一件事就是時間到底以誰為準(zhǔn)。很多人第一次寫 Flink SQL 的時候腦子里只有數(shù)據(jù)來了就處理這一個樸素認(rèn)知于是默認(rèn)什么都不聲明直接SELECT ... FROM kafka_table跑起來發(fā)現(xiàn)結(jié)果時好時壞窗口偶爾觸發(fā)、偶爾不觸發(fā)遲到數(shù)據(jù)要么被丟掉、要么把已經(jīng)算完的結(jié)果又改了一遍。問題往往不在算子邏輯而在于時間語義沒有顯式指定系統(tǒng)只能按默認(rèn)行為走而默認(rèn)行為未必是你業(yè)務(wù)想要的。Flink SQL 里的時間語義落到實(shí)操層面其實(shí)就兩個抓手處理時間Processing Time和事件時間Event Time。處理時間指的是數(shù)據(jù)被算子處理的機(jī)器本地時間簡單、零成本、不需要額外字段但它和數(shù)據(jù)的真實(shí)發(fā)生時刻沒有關(guān)系事件時間指的是數(shù)據(jù)自身攜帶的時間戳反映事件真實(shí)發(fā)生的時刻能正確處理亂序和遲到但代價是需要額外聲明時間屬性和水位線Watermark。這兩個概念在官方文檔里講得很清楚但真正踩過坑的人才知道選錯一個后面所有的窗口、聚合、去重邏輯都會跟著錯。這篇內(nèi)容我打算把這兩條路線徹底講透從 DDL 里怎么聲明、水位線怎么配、窗口怎么綁到實(shí)際跑起來之后會遇到哪些反直覺的現(xiàn)象以及我這些年總結(jié)出來的一套選型判斷標(biāo)準(zhǔn)。適合正在用 Flink SQL 做實(shí)時數(shù)倉、實(shí)時風(fēng)控、實(shí)時監(jiān)控告警的開發(fā)者也適合剛接觸流計算、被時間這個概念繞暈的新手。看完之后你至少能做到拿到一個需求三分鐘內(nèi)判斷該用哪種時間語義并且知道對應(yīng)的 DDL 該怎么寫。2. 處理時間最省事但它的現(xiàn)在不是你的現(xiàn)在2.1 處理時間到底是誰的時間處理時間在 Flink SQL 里通過PROCTIME()函數(shù)聲明它是一個計算列不是數(shù)據(jù)里真實(shí)存在的字段。寫法上通常是這樣CREATE TABLE kafka_source ( user_id BIGINT, item_id BIGINT, behavior STRING, ts BIGINT, proc_time AS PROCTIME() ) WITH ( connector kafka, topic user_behavior, properties.bootstrap.servers localhost:9092, format json );注意proc_time AS PROCTIME()這一行它沒有對應(yīng)的物理字段是 Flink 在運(yùn)行時動態(tài)生成的。它的值等于這條數(shù)據(jù)被當(dāng)前算子處理的機(jī)器系統(tǒng)時間。這里有個非常關(guān)鍵的細(xì)節(jié)處理時間不是數(shù)據(jù)進(jìn)入 Flink 的時間而是數(shù)據(jù)流經(jīng)每一個算子時那個算子所在 TaskManager 的本地時鐘。也就是說同一條數(shù)據(jù)在 source 算子看到的處理時間和在下游窗口算子看到的處理時間理論上可能不一樣雖然差異通常在毫秒級。這個特性帶來一個直接后果處理時間是不可重現(xiàn)的。同一批數(shù)據(jù)你今天跑和明天跑結(jié)果可能不同因?yàn)闄C(jī)器時鐘在變。做離線回刷、做結(jié)果對賬的時候這一點(diǎn)會讓人非常頭疼。我見過有團(tuán)隊(duì)用處理時間做實(shí)時報表結(jié)果某天發(fā)現(xiàn)歷史數(shù)據(jù)對不上排查半天才意識到是處理時間導(dǎo)致的——數(shù)據(jù)重跑時的時間戳全變了。2.2 處理時間窗口為什么看起來很好用處理時間最大的優(yōu)勢是簡單且低延遲。它不需要等待水位線推進(jìn)窗口一旦到達(dá)邊界就立即觸發(fā)計算。比如下面這個滾動窗口SELECT TUMBLE_START(proc_time, INTERVAL 1 MINUTE) AS window_start, item_id, COUNT(*) AS cnt FROM kafka_source GROUP BY TUMBLE(proc_time, INTERVAL 1 MINUTE), item_id;這個查詢的含義是按處理時間每分鐘統(tǒng)計一次每個商品的點(diǎn)擊量。它的行為非常準(zhǔn)時——每到整分鐘窗口就關(guān)閉并輸出結(jié)果。對于監(jiān)控大盤、實(shí)時大屏這類場景這種確定性反而是一種優(yōu)點(diǎn)因?yàn)槟阒澜Y(jié)果什么時候會來。但它的坑也在這里。假設(shè)上游數(shù)據(jù)源因?yàn)榫W(wǎng)絡(luò)抖動積壓了 5 分鐘的數(shù)據(jù)然后一次性灌進(jìn)來。用處理時間的話這 5 分鐘的數(shù)據(jù)會被全部算進(jìn)當(dāng)前這一分鐘的窗口里導(dǎo)致這個窗口的計數(shù)暴漲而前面 5 個窗口的計數(shù)為 0。從業(yè)務(wù)視角看這就是一個典型的數(shù)據(jù)毛刺但根因不在數(shù)據(jù)本身而在時間語義選錯了。2.3 處理時間適用的真實(shí)場景不是說處理時間不能用而是要清楚它的邊界。根據(jù)我的經(jīng)驗(yàn)下面這幾類場景用處理時間是合理的對時間精度要求不高、只關(guān)心最近一段時間的監(jiān)控比如最近 1 分鐘接口錯誤率超過閾值就告警這里最近 1 分鐘指的是系統(tǒng)視角的最近 1 分鐘數(shù)據(jù)本身的亂序不影響告警的時效性。數(shù)據(jù)源本身沒有可靠時間戳有些日志采集鏈路里時間字段要么缺失、要么格式混亂、要么由采集端隨意填充這種情況下強(qiáng)行用事件時間反而會引入更多問題。對結(jié)果可重現(xiàn)性沒有要求的場景比如臨時的數(shù)據(jù)探查、調(diào)試用的統(tǒng)計跑完就丟不需要對賬。反過來只要涉及對賬、計費(fèi)、精確統(tǒng)計、遲到數(shù)據(jù)處理處理時間基本可以直接排除。這不是偏好問題是正確性問題。提示用處理時間時一定要意識到窗口觸發(fā)時刻和數(shù)據(jù)真實(shí)發(fā)生時刻是解耦的。如果你的業(yè)務(wù)方問這個數(shù)字是幾點(diǎn)到幾點(diǎn)的你沒法給出一個讓雙方都滿意的答案。3. 事件時間把時間的定義權(quán)交還給數(shù)據(jù)本身3.1 事件時間字段與水位線的綁定關(guān)系事件時間的核心是數(shù)據(jù)自帶時間戳加上水位線。時間戳告訴系統(tǒng)這件事是什么時候發(fā)生的水位線告訴系統(tǒng)我認(rèn)為比某個時刻更早的數(shù)據(jù)都已經(jīng)到齊了。兩者缺一不可。在 Flink SQL 的 DDL 里事件時間的聲明方式是這樣CREATE TABLE kafka_source ( user_id BIGINT, item_id BIGINT, behavior STRING, ts BIGINT, row_time AS TO_TIMESTAMP_LTZ(ts, 3), WATERMARK FOR row_time AS row_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic user_behavior, properties.bootstrap.servers localhost:9092, format json );這里有幾個點(diǎn)必須掰開講。第一row_time是一個計算列通過TO_TIMESTAMP_LTZ把毫秒時間戳轉(zhuǎn)成 Flink 內(nèi)部的TIMESTAMP_LTZ類型。為什么用LTZ帶本地時區(qū)而不是普通TIMESTAMP因?yàn)門IMESTAMP不帶時區(qū)信息在做跨時區(qū)或者和外部系統(tǒng)交互時容易出問題TIMESTAMP_LTZ更穩(wěn)妥。第二WATERMARK FOR row_time AS row_time - INTERVAL 5 SECOND定義了水位線生成策略水位線 當(dāng)前觀察到的最大事件時間 - 5 秒。這個 5 秒就是允許的亂序程度。水位線的本質(zhì)是一個單調(diào)遞增的時間戳它只會往前走不會后退。當(dāng)水位線推進(jìn)到某個窗口的結(jié)束時間時系統(tǒng)就認(rèn)為這個窗口的數(shù)據(jù)都到齊了可以觸發(fā)計算。所以水位線的推進(jìn)速度直接決定了窗口的觸發(fā)時機(jī)和延遲。3.2 水位線延遲設(shè)多少才合適這是被問得最多的問題之一。設(shè)小了遲到數(shù)據(jù)被丟設(shè)大了窗口遲遲不觸發(fā)端到端延遲高。我的經(jīng)驗(yàn)是分三步走第一步先看數(shù)據(jù)的實(shí)際亂序分布。把上游數(shù)據(jù)抽樣出來統(tǒng)計每條數(shù)據(jù)的事件時間和它到達(dá)系統(tǒng)的時間之差畫一個分布圖。如果 99% 的數(shù)據(jù)延遲都在 3 秒以內(nèi)那水位線設(shè) 5 秒是比較安全的如果長尾能到 30 秒那就要考慮設(shè) 30 秒甚至更長。第二步結(jié)合業(yè)務(wù)對延遲的容忍度。實(shí)時風(fēng)控可能要求秒級出結(jié)果那水位線就不能設(shè)太大寧可接受少量遲到數(shù)據(jù)被丟棄或者用側(cè)輸出流單獨(dú)處理。實(shí)時報表對延遲容忍度高一些水位線可以設(shè)大一點(diǎn)保證數(shù)據(jù)完整性。第三步上線后持續(xù)觀察。Flink 的 Web UI 里能看到水位線的推進(jìn)情況和窗口的觸發(fā)延遲配合監(jiān)控指標(biāo)比如currentWatermark、numLateRecordsDropped可以判斷當(dāng)前設(shè)置是否合理。如果numLateRecordsDropped持續(xù)大于 0說明水位線設(shè)小了如果窗口觸發(fā)延遲遠(yuǎn)大于業(yè)務(wù)容忍度說明設(shè)大了。下面這張表是我在不同業(yè)務(wù)場景下的經(jīng)驗(yàn)值僅供參考業(yè)務(wù)場景典型亂序程度建議水位線延遲備注實(shí)時監(jiān)控告警秒級3-5 秒優(yōu)先保證時效性實(shí)時數(shù)倉明細(xì)秒級到十秒級10-30 秒平衡完整性和延遲實(shí)時計費(fèi)對賬分鐘級1-5 分鐘優(yōu)先保證完整性日志分析不確定先采樣再定必須做數(shù)據(jù)探查3.3 事件時間窗口的觸發(fā)邏輯用事件時間做窗口寫法上和處理時間很像只是把PROCTIME()換成事件時間字段SELECT TUMBLE_START(row_time, INTERVAL 1 MINUTE) AS window_start, item_id, COUNT(*) AS cnt FROM kafka_source GROUP BY TUMBLE(row_time, INTERVAL 1 MINUTE), item_id;但行為完全不同。這個查詢不會在整分鐘觸發(fā)而是等水位線越過窗口結(jié)束時間才觸發(fā)。假設(shè)水位線延遲是 5 秒那么 10:00:00 到 10:01:00 這個窗口會在系統(tǒng)觀察到事件時間超過 10:01:05 的數(shù)據(jù)時觸發(fā)。這意味著窗口的輸出是滯后的滯后程度約等于水位線延遲。這個滯后是必要的代價。正是這段等待時間讓亂序到達(dá)的數(shù)據(jù)有機(jī)會被正確歸入它所屬的窗口。如果一條數(shù)據(jù)的事件時間是 10:00:30但它在 10:01:10 才到達(dá)只要水位線還沒越過 10:01:00它依然會被算進(jìn)正確的窗口。這就是事件時間的價值所在。4. 兩種時間語義混用時的邊界與陷阱4.1 同一個查詢里能不能同時用兩種時間可以但要非常小心。Flink SQL 允許在一個查詢里同時存在處理時間和事件時間字段比如一張表既有proc_time AS PROCTIME()又有row_time和對應(yīng)的水位線。但一個窗口操作只能綁定一種時間屬性你不能讓一個TUMBLE同時按兩種時間切分。更隱蔽的坑在于雙流 JOIN。當(dāng)兩張表按時間窗口做 JOIN 時如果一張用處理時間、一張用事件時間Flink 會報錯或者行為不符合預(yù)期。正確的做法是兩邊統(tǒng)一。我見過一個案例訂單流用事件時間支付流用處理時間JOIN 出來的結(jié)果時對時錯排查了兩天才定位到是時間語義不一致。-- 錯誤示范兩邊時間語義不一致 SELECT ... FROM order_stream o JOIN payment_stream p ON o.order_id p.order_id AND o.row_time BETWEEN p.proc_time - INTERVAL 5 MINUTE AND p.proc_time INTERVAL 5 MINUTE;這種寫法即使能跑語義也是混亂的。正確做法是讓兩張表都用事件時間并且水位線策略盡量對齊。4.2 時間屬性的傳遞規(guī)則還有一個容易被忽略的點(diǎn)時間屬性在算子間的傳遞。在 Flink SQL 里時間屬性尤其是事件時間經(jīng)過某些算子后會丟失。比如經(jīng)過GROUP BY聚合之后輸出的結(jié)果里就不再帶有原始的時間屬性和水位線了。如果你在聚合結(jié)果上再做窗口會發(fā)現(xiàn)沒有可用的時間字段。解決辦法通常有兩種一是把窗口時間作為普通字段顯式 SELECT 出來比如TUMBLE_START的結(jié)果后續(xù)基于這個字段做邏輯二是在聚合時保留時間屬性但這需要看具體算子的支持情況。我的建議是盡量把窗口操作放在靠近數(shù)據(jù)源的位置避免在多層聚合之后再套窗口那樣既難維護(hù)又容易出錯。4.3 遲到數(shù)據(jù)的三種處理策略事件時間雖然能容忍亂序但容忍是有限度的。超過水位線才到達(dá)的數(shù)據(jù)就是遲到數(shù)據(jù)默認(rèn)行為是被丟棄。如果你不能接受丟棄Flink SQL 提供了幾種策略允許窗口延遲關(guān)閉通過table.exec.emit.allow-late-fire之類的配置不同版本參數(shù)名可能不同讓窗口在水位線越過后仍然接受一段時間的遲到數(shù)據(jù)并重新觸發(fā)計算。代價是結(jié)果會被更新下游需要能處理更新流。側(cè)輸出流把遲到數(shù)據(jù)單獨(dú)輸出到一個側(cè)流后續(xù)用批處理或者人工方式補(bǔ)償。這是最靈活的方式但需要額外的代碼。加大水位線延遲最簡單粗暴但會犧牲端到端延遲而且治標(biāo)不治本。選擇哪種取決于業(yè)務(wù)對數(shù)據(jù)完整性和結(jié)果確定性的要求。計費(fèi)類業(yè)務(wù)通常選側(cè)輸出流因?yàn)槊恳还P都不能丟監(jiān)控類業(yè)務(wù)通常直接丟棄因?yàn)檫t到的那一條對整體趨勢影響不大。5. 從 DDL 到窗口一套可復(fù)現(xiàn)的配置流程5.1 環(huán)境與版本確認(rèn)在動手之前先確認(rèn)你的 Flink 版本。不同版本在時間屬性的語法上有細(xì)微差異尤其是TO_TIMESTAMP_LTZ這個函數(shù)是 1.13 之后才穩(wěn)定的。我一般會先跑一句SELECT VERSION()確認(rèn)避免寫完 DDL 才發(fā)現(xiàn)函數(shù)不存在。另外如果用 Kafka 作為源format的選擇也會影響時間字段的解析。JSON 格式下時間戳通常是數(shù)字需要顯式轉(zhuǎn)換如果用 CDC 格式時間字段的處理方式又不一樣。這些細(xì)節(jié)在寫 DDL 時就要考慮進(jìn)去。5.2 完整 DDL 與窗口查詢示例下面給一套完整可跑的配置以事件時間為例-- 1. 建源表聲明事件時間和水位線 CREATE TABLE user_behavior ( user_id BIGINT, item_id BIGINT, behavior STRING, ts BIGINT, row_time AS TO_TIMESTAMP_LTZ(ts, 3), WATERMARK FOR row_time AS row_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic user_behavior, properties.bootstrap.servers localhost:9092, properties.group.id test_group, scan.startup.mode latest-offset, format json ); -- 2. 建結(jié)果表 CREATE TABLE item_cnt_result ( window_start TIMESTAMP(3), item_id BIGINT, cnt BIGINT ) WITH ( connector kafka, topic item_cnt_result, properties.bootstrap.servers localhost:9092, format json ); -- 3. 窗口聚合寫入 INSERT INTO item_cnt_result SELECT TUMBLE_START(row_time, INTERVAL 1 MINUTE) AS window_start, item_id, COUNT(*) AS cnt FROM user_behavior GROUP BY TUMBLE(row_time, INTERVAL 1 MINUTE), item_id;這套配置跑起來之后你可以通過 Flink Web UI 觀察水位線的推進(jìn)。如果發(fā)現(xiàn)水位線長時間不動通常是某個分區(qū)沒有數(shù)據(jù)或者數(shù)據(jù)的時間戳格式解析失敗導(dǎo)致水位線無法推進(jìn)。5.3 驗(yàn)證時間語義是否生效寫完不代表對了。我通常會做三個驗(yàn)證第一造一批亂序數(shù)據(jù)。手動往 Kafka 里寫幾條事件時間亂序的數(shù)據(jù)看它們是否被正確歸入對應(yīng)窗口。比如先寫 10:00:30 的再寫 10:00:10 的如果兩條都進(jìn)了 10:00 這個窗口說明事件時間生效了。第二觀察遲到數(shù)據(jù)。寫一條事件時間遠(yuǎn)早于當(dāng)前水位線的數(shù)據(jù)看它是否被丟棄以及numLateRecordsDropped指標(biāo)是否增加。第三對比處理時間。同樣的數(shù)據(jù)分別用處理時間和事件時間跑一遍對比結(jié)果差異。如果差異很大說明數(shù)據(jù)亂序嚴(yán)重事件時間是必要的如果幾乎沒差異說明數(shù)據(jù)基本有序處理時間也能湊合。6. 那些文檔里不會寫的實(shí)操心得6.1 水位線不是越大越好新手容易有一個誤區(qū)既然水位線大能容忍更多亂序那干脆設(shè)個 10 分鐘。結(jié)果發(fā)現(xiàn)端到端延遲高得離譜業(yè)務(wù)方天天催。水位線的本質(zhì)是用延遲換完整性這個 trade-off 必須結(jié)合業(yè)務(wù)來定。我的做法是先設(shè)一個保守值比如 5 秒上線后看遲到數(shù)據(jù)比例如果低于千分之一就不動如果高于百分之一再考慮調(diào)整。6.2 時間戳字段的時區(qū)問題這是最隱蔽的坑之一。如果上游數(shù)據(jù)的時間戳是秒級或毫秒級數(shù)字用TO_TIMESTAMP_LTZ轉(zhuǎn)換時Flink 會按本地時區(qū)解釋。如果上游用的是 UTC 時間而你的 Flink 集群時區(qū)是東八區(qū)轉(zhuǎn)換出來的時間就會差 8 小時。解決辦法是在 DDL 里顯式指定時區(qū)或者在轉(zhuǎn)換前先做偏移。這個問題在跨時區(qū)部署時尤其常見我踩過不止一次。6.3 空閑分區(qū)導(dǎo)致水位線卡住如果 Kafka 有多個分區(qū)而某個分區(qū)長時間沒有數(shù)據(jù)Flink 默認(rèn)會認(rèn)為這個分區(qū)的水位線不推進(jìn)從而導(dǎo)致整個算子的水位線卡住。表現(xiàn)就是窗口遲遲不觸發(fā)。解決辦法是開啟空閑分區(qū)檢測-- 在 DDL 的 WITH 參數(shù)里加上 table.exec.source.idle-timeout 30s這個配置的意思是如果一個分區(qū) 30 秒沒有數(shù)據(jù)就把它標(biāo)記為空閑水位線計算時忽略它。這個參數(shù)在 Flink 1.11 之后可用非常實(shí)用建議默認(rèn)加上。6.4 窗口結(jié)果更新與下游兼容用事件時間加允許遲到數(shù)據(jù)時窗口結(jié)果可能會被更新retract 流。如果下游是 Kafka 這種只追加的系統(tǒng)更新會以新消息的形式發(fā)出下游需要能處理這種同 key 多條消息的情況。如果下游是數(shù)據(jù)庫通常需要按主鍵做 upsert。這一點(diǎn)在設(shè)計鏈路時就要想清楚否則上線后會發(fā)現(xiàn)結(jié)果表里有重復(fù)數(shù)據(jù)。7. 選型決策三分鐘判斷該用哪種時間把前面的內(nèi)容濃縮成一個決策流程實(shí)際工作中我基本按這個走第一步問數(shù)據(jù)源有沒有可靠的事件時間戳。沒有的話直接處理時間別糾結(jié)。有的話進(jìn)入第二步。第二步問業(yè)務(wù)能不能接受結(jié)果不可重現(xiàn)。涉及對賬、計費(fèi)、審計的不能接受必須事件時間。純監(jiān)控、純展示的可以接受處理時間更簡單。第三步問業(yè)務(wù)對延遲的容忍度。如果要求秒級出結(jié)果而數(shù)據(jù)亂序又很嚴(yán)重那就要權(quán)衡要么接受少量數(shù)據(jù)丟失用事件時間加小水位線要么用處理時間接受結(jié)果偏差。第四步問團(tuán)隊(duì)對水位線的運(yùn)維能力。事件時間需要監(jiān)控水位線推進(jìn)、遲到數(shù)據(jù)比例、窗口觸發(fā)延遲運(yùn)維成本更高。如果團(tuán)隊(duì)沒有相應(yīng)的監(jiān)控體系處理時間是更穩(wěn)妥的選擇。這四步走下來基本能覆蓋 90% 的場景。剩下的 10%通常是雙流 JOIN、多維聚合這類復(fù)雜場景需要具體問題具體分析但核心判斷邏輯是一樣的時間是數(shù)據(jù)的屬性還是系統(tǒng)的屬性決定了你用哪種語義。最后分享一個我自己的習(xí)慣任何涉及時間的 Flink SQL 作業(yè)上線前我都會在 DDL 旁邊寫一行注釋說明為什么選這種時間語義、水位線為什么設(shè)這個值。過幾個月回頭看這行注釋能省下大量重新推理的時間。時間語義這東西寫的時候覺得理所當(dāng)然改的時候才發(fā)現(xiàn)處處是坑。 Flink SQL時間語義選型指南：處理時間與事件時間實(shí)戰(zhàn)
1. 時間語義選錯整條實(shí)時鏈路都在白跑實(shí)時計算里最容易被低估的一件事就是時間到底以誰為準(zhǔn)。很多人第一次寫 Flink SQL 的時候腦子里只有數(shù)據(jù)來了就處理這一個樸素認(rèn)知于是默認(rèn)什么都不聲明直接SELECT ... FROM kafka_table跑起來發(fā)現(xiàn)結(jié)果時好時壞窗口偶爾觸發(fā)、偶爾不觸發(fā)遲到數(shù)據(jù)要么被丟掉、要么把已經(jīng)算完的結(jié)果又改了一遍。問題往往不在算子邏輯而在于時間語義沒有顯式指定系統(tǒng)只能按默認(rèn)行為走而默認(rèn)行為未必是你業(yè)務(wù)想要的。Flink SQL 里的時間語義落到實(shí)操層面其實(shí)就兩個抓手處理時間Processing Time和事件時間Event Time。處理時間指的是數(shù)據(jù)被算子處理的機(jī)器本地時間簡單、零成本、不需要額外字段但它和數(shù)據(jù)的真實(shí)發(fā)生時刻沒有關(guān)系事件時間指的是數(shù)據(jù)自身攜帶的時間戳反映事件真實(shí)發(fā)生的時刻能正確處理亂序和遲到但代價是需要額外聲明時間屬性和水位線Watermark。這兩個概念在官方文檔里講得很清楚但真正踩過坑的人才知道選錯一個后面所有的窗口、聚合、去重邏輯都會跟著錯。這篇內(nèi)容我打算把這兩條路線徹底講透從 DDL 里怎么聲明、水位線怎么配、窗口怎么綁到實(shí)際跑起來之后會遇到哪些反直覺的現(xiàn)象以及我這些年總結(jié)出來的一套選型判斷標(biāo)準(zhǔn)。適合正在用 Flink SQL 做實(shí)時數(shù)倉、實(shí)時風(fēng)控、實(shí)時監(jiān)控告警的開發(fā)者也適合剛接觸流計算、被時間這個概念繞暈的新手。看完之后你至少能做到拿到一個需求三分鐘內(nèi)判斷該用哪種時間語義并且知道對應(yīng)的 DDL 該怎么寫。2. 處理時間最省事但它的現(xiàn)在不是你的現(xiàn)在2.1 處理時間到底是誰的時間處理時間在 Flink SQL 里通過PROCTIME()函數(shù)聲明它是一個計算列不是數(shù)據(jù)里真實(shí)存在的字段。寫法上通常是這樣CREATE TABLE kafka_source ( user_id BIGINT, item_id BIGINT, behavior STRING, ts BIGINT, proc_time AS PROCTIME() ) WITH ( connector kafka, topic user_behavior, properties.bootstrap.servers localhost:9092, format json );注意proc_time AS PROCTIME()這一行它沒有對應(yīng)的物理字段是 Flink 在運(yùn)行時動態(tài)生成的。它的值等于這條數(shù)據(jù)被當(dāng)前算子處理的機(jī)器系統(tǒng)時間。這里有個非常關(guān)鍵的細(xì)節(jié)處理時間不是數(shù)據(jù)進(jìn)入 Flink 的時間而是數(shù)據(jù)流經(jīng)每一個算子時那個算子所在 TaskManager 的本地時鐘。也就是說同一條數(shù)據(jù)在 source 算子看到的處理時間和在下游窗口算子看到的處理時間理論上可能不一樣雖然差異通常在毫秒級。這個特性帶來一個直接后果處理時間是不可重現(xiàn)的。同一批數(shù)據(jù)你今天跑和明天跑結(jié)果可能不同因?yàn)闄C(jī)器時鐘在變。做離線回刷、做結(jié)果對賬的時候這一點(diǎn)會讓人非常頭疼。我見過有團(tuán)隊(duì)用處理時間做實(shí)時報表結(jié)果某天發(fā)現(xiàn)歷史數(shù)據(jù)對不上排查半天才意識到是處理時間導(dǎo)致的——數(shù)據(jù)重跑時的時間戳全變了。2.2 處理時間窗口為什么看起來很好用處理時間最大的優(yōu)勢是簡單且低延遲。它不需要等待水位線推進(jìn)窗口一旦到達(dá)邊界就立即觸發(fā)計算。比如下面這個滾動窗口SELECT TUMBLE_START(proc_time, INTERVAL 1 MINUTE) AS window_start, item_id, COUNT(*) AS cnt FROM kafka_source GROUP BY TUMBLE(proc_time, INTERVAL 1 MINUTE), item_id;這個查詢的含義是按處理時間每分鐘統(tǒng)計一次每個商品的點(diǎn)擊量。它的行為非常準(zhǔn)時——每到整分鐘窗口就關(guān)閉并輸出結(jié)果。對于監(jiān)控大盤、實(shí)時大屏這類場景這種確定性反而是一種優(yōu)點(diǎn)因?yàn)槟阒澜Y(jié)果什么時候會來。但它的坑也在這里。假設(shè)上游數(shù)據(jù)源因?yàn)榫W(wǎng)絡(luò)抖動積壓了 5 分鐘的數(shù)據(jù)然后一次性灌進(jìn)來。用處理時間的話這 5 分鐘的數(shù)據(jù)會被全部算進(jìn)當(dāng)前這一分鐘的窗口里導(dǎo)致這個窗口的計數(shù)暴漲而前面 5 個窗口的計數(shù)為 0。從業(yè)務(wù)視角看這就是一個典型的數(shù)據(jù)毛刺但根因不在數(shù)據(jù)本身而在時間語義選錯了。2.3 處理時間適用的真實(shí)場景不是說處理時間不能用而是要清楚它的邊界。根據(jù)我的經(jīng)驗(yàn)下面這幾類場景用處理時間是合理的對時間精度要求不高、只關(guān)心最近一段時間的監(jiān)控比如最近 1 分鐘接口錯誤率超過閾值就告警這里最近 1 分鐘指的是系統(tǒng)視角的最近 1 分鐘數(shù)據(jù)本身的亂序不影響告警的時效性。數(shù)據(jù)源本身沒有可靠時間戳有些日志采集鏈路里時間字段要么缺失、要么格式混亂、要么由采集端隨意填充這種情況下強(qiáng)行用事件時間反而會引入更多問題。對結(jié)果可重現(xiàn)性沒有要求的場景比如臨時的數(shù)據(jù)探查、調(diào)試用的統(tǒng)計跑完就丟不需要對賬。反過來只要涉及對賬、計費(fèi)、精確統(tǒng)計、遲到數(shù)據(jù)處理處理時間基本可以直接排除。這不是偏好問題是正確性問題。提示用處理時間時一定要意識到窗口觸發(fā)時刻和數(shù)據(jù)真實(shí)發(fā)生時刻是解耦的。如果你的業(yè)務(wù)方問這個數(shù)字是幾點(diǎn)到幾點(diǎn)的你沒法給出一個讓雙方都滿意的答案。3. 事件時間把時間的定義權(quán)交還給數(shù)據(jù)本身3.1 事件時間字段與水位線的綁定關(guān)系事件時間的核心是數(shù)據(jù)自帶時間戳加上水位線。時間戳告訴系統(tǒng)這件事是什么時候發(fā)生的水位線告訴系統(tǒng)我認(rèn)為比某個時刻更早的數(shù)據(jù)都已經(jīng)到齊了。兩者缺一不可。在 Flink SQL 的 DDL 里事件時間的聲明方式是這樣CREATE TABLE kafka_source ( user_id BIGINT, item_id BIGINT, behavior STRING, ts BIGINT, row_time AS TO_TIMESTAMP_LTZ(ts, 3), WATERMARK FOR row_time AS row_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic user_behavior, properties.bootstrap.servers localhost:9092, format json );這里有幾個點(diǎn)必須掰開講。第一row_time是一個計算列通過TO_TIMESTAMP_LTZ把毫秒時間戳轉(zhuǎn)成 Flink 內(nèi)部的TIMESTAMP_LTZ類型。為什么用LTZ帶本地時區(qū)而不是普通TIMESTAMP因?yàn)門IMESTAMP不帶時區(qū)信息在做跨時區(qū)或者和外部系統(tǒng)交互時容易出問題TIMESTAMP_LTZ更穩(wěn)妥。第二WATERMARK FOR row_time AS row_time - INTERVAL 5 SECOND定義了水位線生成策略水位線 當(dāng)前觀察到的最大事件時間 - 5 秒。這個 5 秒就是允許的亂序程度。水位線的本質(zhì)是一個單調(diào)遞增的時間戳它只會往前走不會后退。當(dāng)水位線推進(jìn)到某個窗口的結(jié)束時間時系統(tǒng)就認(rèn)為這個窗口的數(shù)據(jù)都到齊了可以觸發(fā)計算。所以水位線的推進(jìn)速度直接決定了窗口的觸發(fā)時機(jī)和延遲。3.2 水位線延遲設(shè)多少才合適這是被問得最多的問題之一。設(shè)小了遲到數(shù)據(jù)被丟設(shè)大了窗口遲遲不觸發(fā)端到端延遲高。我的經(jīng)驗(yàn)是分三步走第一步先看數(shù)據(jù)的實(shí)際亂序分布。把上游數(shù)據(jù)抽樣出來統(tǒng)計每條數(shù)據(jù)的事件時間和它到達(dá)系統(tǒng)的時間之差畫一個分布圖。如果 99% 的數(shù)據(jù)延遲都在 3 秒以內(nèi)那水位線設(shè) 5 秒是比較安全的如果長尾能到 30 秒那就要考慮設(shè) 30 秒甚至更長。第二步結(jié)合業(yè)務(wù)對延遲的容忍度。實(shí)時風(fēng)控可能要求秒級出結(jié)果那水位線就不能設(shè)太大寧可接受少量遲到數(shù)據(jù)被丟棄或者用側(cè)輸出流單獨(dú)處理。實(shí)時報表對延遲容忍度高一些水位線可以設(shè)大一點(diǎn)保證數(shù)據(jù)完整性。第三步上線后持續(xù)觀察。Flink 的 Web UI 里能看到水位線的推進(jìn)情況和窗口的觸發(fā)延遲配合監(jiān)控指標(biāo)比如currentWatermark、numLateRecordsDropped可以判斷當(dāng)前設(shè)置是否合理。如果numLateRecordsDropped持續(xù)大于 0說明水位線設(shè)小了如果窗口觸發(fā)延遲遠(yuǎn)大于業(yè)務(wù)容忍度說明設(shè)大了。下面這張表是我在不同業(yè)務(wù)場景下的經(jīng)驗(yàn)值僅供參考業(yè)務(wù)場景典型亂序程度建議水位線延遲備注實(shí)時監(jiān)控告警秒級3-5 秒優(yōu)先保證時效性實(shí)時數(shù)倉明細(xì)秒級到十秒級10-30 秒平衡完整性和延遲實(shí)時計費(fèi)對賬分鐘級1-5 分鐘優(yōu)先保證完整性日志分析不確定先采樣再定必須做數(shù)據(jù)探查3.3 事件時間窗口的觸發(fā)邏輯用事件時間做窗口寫法上和處理時間很像只是把PROCTIME()換成事件時間字段SELECT TUMBLE_START(row_time, INTERVAL 1 MINUTE) AS window_start, item_id, COUNT(*) AS cnt FROM kafka_source GROUP BY TUMBLE(row_time, INTERVAL 1 MINUTE), item_id;但行為完全不同。這個查詢不會在整分鐘觸發(fā)而是等水位線越過窗口結(jié)束時間才觸發(fā)。假設(shè)水位線延遲是 5 秒那么 10:00:00 到 10:01:00 這個窗口會在系統(tǒng)觀察到事件時間超過 10:01:05 的數(shù)據(jù)時觸發(fā)。這意味著窗口的輸出是滯后的滯后程度約等于水位線延遲。這個滯后是必要的代價。正是這段等待時間讓亂序到達(dá)的數(shù)據(jù)有機(jī)會被正確歸入它所屬的窗口。如果一條數(shù)據(jù)的事件時間是 10:00:30但它在 10:01:10 才到達(dá)只要水位線還沒越過 10:01:00它依然會被算進(jìn)正確的窗口。這就是事件時間的價值所在。4. 兩種時間語義混用時的邊界與陷阱4.1 同一個查詢里能不能同時用兩種時間可以但要非常小心。Flink SQL 允許在一個查詢里同時存在處理時間和事件時間字段比如一張表既有proc_time AS PROCTIME()又有row_time和對應(yīng)的水位線。但一個窗口操作只能綁定一種時間屬性你不能讓一個TUMBLE同時按兩種時間切分。更隱蔽的坑在于雙流 JOIN。當(dāng)兩張表按時間窗口做 JOIN 時如果一張用處理時間、一張用事件時間Flink 會報錯或者行為不符合預(yù)期。正確的做法是兩邊統(tǒng)一。我見過一個案例訂單流用事件時間支付流用處理時間JOIN 出來的結(jié)果時對時錯排查了兩天才定位到是時間語義不一致。-- 錯誤示范兩邊時間語義不一致 SELECT ... FROM order_stream o JOIN payment_stream p ON o.order_id p.order_id AND o.row_time BETWEEN p.proc_time - INTERVAL 5 MINUTE AND p.proc_time INTERVAL 5 MINUTE;這種寫法即使能跑語義也是混亂的。正確做法是讓兩張表都用事件時間并且水位線策略盡量對齊。4.2 時間屬性的傳遞規(guī)則還有一個容易被忽略的點(diǎn)時間屬性在算子間的傳遞。在 Flink SQL 里時間屬性尤其是事件時間經(jīng)過某些算子后會丟失。比如經(jīng)過GROUP BY聚合之后輸出的結(jié)果里就不再帶有原始的時間屬性和水位線了。如果你在聚合結(jié)果上再做窗口會發(fā)現(xiàn)沒有可用的時間字段。解決辦法通常有兩種一是把窗口時間作為普通字段顯式 SELECT 出來比如TUMBLE_START的結(jié)果后續(xù)基于這個字段做邏輯二是在聚合時保留時間屬性但這需要看具體算子的支持情況。我的建議是盡量把窗口操作放在靠近數(shù)據(jù)源的位置避免在多層聚合之后再套窗口那樣既難維護(hù)又容易出錯。4.3 遲到數(shù)據(jù)的三種處理策略事件時間雖然能容忍亂序但容忍是有限度的。超過水位線才到達(dá)的數(shù)據(jù)就是遲到數(shù)據(jù)默認(rèn)行為是被丟棄。如果你不能接受丟棄Flink SQL 提供了幾種策略允許窗口延遲關(guān)閉通過table.exec.emit.allow-late-fire之類的配置不同版本參數(shù)名可能不同讓窗口在水位線越過后仍然接受一段時間的遲到數(shù)據(jù)并重新觸發(fā)計算。代價是結(jié)果會被更新下游需要能處理更新流。側(cè)輸出流把遲到數(shù)據(jù)單獨(dú)輸出到一個側(cè)流后續(xù)用批處理或者人工方式補(bǔ)償。這是最靈活的方式但需要額外的代碼。加大水位線延遲最簡單粗暴但會犧牲端到端延遲而且治標(biāo)不治本。選擇哪種取決于業(yè)務(wù)對數(shù)據(jù)完整性和結(jié)果確定性的要求。計費(fèi)類業(yè)務(wù)通常選側(cè)輸出流因?yàn)槊恳还P都不能丟監(jiān)控類業(yè)務(wù)通常直接丟棄因?yàn)檫t到的那一條對整體趨勢影響不大。5. 從 DDL 到窗口一套可復(fù)現(xiàn)的配置流程5.1 環(huán)境與版本確認(rèn)在動手之前先確認(rèn)你的 Flink 版本。不同版本在時間屬性的語法上有細(xì)微差異尤其是TO_TIMESTAMP_LTZ這個函數(shù)是 1.13 之后才穩(wěn)定的。我一般會先跑一句SELECT VERSION()確認(rèn)避免寫完 DDL 才發(fā)現(xiàn)函數(shù)不存在。另外如果用 Kafka 作為源format的選擇也會影響時間字段的解析。JSON 格式下時間戳通常是數(shù)字需要顯式轉(zhuǎn)換如果用 CDC 格式時間字段的處理方式又不一樣。這些細(xì)節(jié)在寫 DDL 時就要考慮進(jìn)去。5.2 完整 DDL 與窗口查詢示例下面給一套完整可跑的配置以事件時間為例-- 1. 建源表聲明事件時間和水位線 CREATE TABLE user_behavior ( user_id BIGINT, item_id BIGINT, behavior STRING, ts BIGINT, row_time AS TO_TIMESTAMP_LTZ(ts, 3), WATERMARK FOR row_time AS row_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic user_behavior, properties.bootstrap.servers localhost:9092, properties.group.id test_group, scan.startup.mode latest-offset, format json ); -- 2. 建結(jié)果表 CREATE TABLE item_cnt_result ( window_start TIMESTAMP(3), item_id BIGINT, cnt BIGINT ) WITH ( connector kafka, topic item_cnt_result, properties.bootstrap.servers localhost:9092, format json ); -- 3. 窗口聚合寫入 INSERT INTO item_cnt_result SELECT TUMBLE_START(row_time, INTERVAL 1 MINUTE) AS window_start, item_id, COUNT(*) AS cnt FROM user_behavior GROUP BY TUMBLE(row_time, INTERVAL 1 MINUTE), item_id;這套配置跑起來之后你可以通過 Flink Web UI 觀察水位線的推進(jìn)。如果發(fā)現(xiàn)水位線長時間不動通常是某個分區(qū)沒有數(shù)據(jù)或者數(shù)據(jù)的時間戳格式解析失敗導(dǎo)致水位線無法推進(jìn)。5.3 驗(yàn)證時間語義是否生效寫完不代表對了。我通常會做三個驗(yàn)證第一造一批亂序數(shù)據(jù)。手動往 Kafka 里寫幾條事件時間亂序的數(shù)據(jù)看它們是否被正確歸入對應(yīng)窗口。比如先寫 10:00:30 的再寫 10:00:10 的如果兩條都進(jìn)了 10:00 這個窗口說明事件時間生效了。第二觀察遲到數(shù)據(jù)。寫一條事件時間遠(yuǎn)早于當(dāng)前水位線的數(shù)據(jù)看它是否被丟棄以及numLateRecordsDropped指標(biāo)是否增加。第三對比處理時間。同樣的數(shù)據(jù)分別用處理時間和事件時間跑一遍對比結(jié)果差異。如果差異很大說明數(shù)據(jù)亂序嚴(yán)重事件時間是必要的如果幾乎沒差異說明數(shù)據(jù)基本有序處理時間也能湊合。6. 那些文檔里不會寫的實(shí)操心得6.1 水位線不是越大越好新手容易有一個誤區(qū)既然水位線大能容忍更多亂序那干脆設(shè)個 10 分鐘。結(jié)果發(fā)現(xiàn)端到端延遲高得離譜業(yè)務(wù)方天天催。水位線的本質(zhì)是用延遲換完整性這個 trade-off 必須結(jié)合業(yè)務(wù)來定。我的做法是先設(shè)一個保守值比如 5 秒上線后看遲到數(shù)據(jù)比例如果低于千分之一就不動如果高于百分之一再考慮調(diào)整。6.2 時間戳字段的時區(qū)問題這是最隱蔽的坑之一。如果上游數(shù)據(jù)的時間戳是秒級或毫秒級數(shù)字用TO_TIMESTAMP_LTZ轉(zhuǎn)換時Flink 會按本地時區(qū)解釋。如果上游用的是 UTC 時間而你的 Flink 集群時區(qū)是東八區(qū)轉(zhuǎn)換出來的時間就會差 8 小時。解決辦法是在 DDL 里顯式指定時區(qū)或者在轉(zhuǎn)換前先做偏移。這個問題在跨時區(qū)部署時尤其常見我踩過不止一次。6.3 空閑分區(qū)導(dǎo)致水位線卡住如果 Kafka 有多個分區(qū)而某個分區(qū)長時間沒有數(shù)據(jù)Flink 默認(rèn)會認(rèn)為這個分區(qū)的水位線不推進(jìn)從而導(dǎo)致整個算子的水位線卡住。表現(xiàn)就是窗口遲遲不觸發(fā)。解決辦法是開啟空閑分區(qū)檢測-- 在 DDL 的 WITH 參數(shù)里加上 table.exec.source.idle-timeout 30s這個配置的意思是如果一個分區(qū) 30 秒沒有數(shù)據(jù)就把它標(biāo)記為空閑水位線計算時忽略它。這個參數(shù)在 Flink 1.11 之后可用非常實(shí)用建議默認(rèn)加上。6.4 窗口結(jié)果更新與下游兼容用事件時間加允許遲到數(shù)據(jù)時窗口結(jié)果可能會被更新retract 流。如果下游是 Kafka 這種只追加的系統(tǒng)更新會以新消息的形式發(fā)出下游需要能處理這種同 key 多條消息的情況。如果下游是數(shù)據(jù)庫通常需要按主鍵做 upsert。這一點(diǎn)在設(shè)計鏈路時就要想清楚否則上線后會發(fā)現(xiàn)結(jié)果表里有重復(fù)數(shù)據(jù)。7. 選型決策三分鐘判斷該用哪種時間把前面的內(nèi)容濃縮成一個決策流程實(shí)際工作中我基本按這個走第一步問數(shù)據(jù)源有沒有可靠的事件時間戳。沒有的話直接處理時間別糾結(jié)。有的話進(jìn)入第二步。第二步問業(yè)務(wù)能不能接受結(jié)果不可重現(xiàn)。涉及對賬、計費(fèi)、審計的不能接受必須事件時間。純監(jiān)控、純展示的可以接受處理時間更簡單。第三步問業(yè)務(wù)對延遲的容忍度。如果要求秒級出結(jié)果而數(shù)據(jù)亂序又很嚴(yán)重那就要權(quán)衡要么接受少量數(shù)據(jù)丟失用事件時間加小水位線要么用處理時間接受結(jié)果偏差。第四步問團(tuán)隊(duì)對水位線的運(yùn)維能力。事件時間需要監(jiān)控水位線推進(jìn)、遲到數(shù)據(jù)比例、窗口觸發(fā)延遲運(yùn)維成本更高。如果團(tuán)隊(duì)沒有相應(yīng)的監(jiān)控體系處理時間是更穩(wěn)妥的選擇。這四步走下來基本能覆蓋 90% 的場景。剩下的 10%通常是雙流 JOIN、多維聚合這類復(fù)雜場景需要具體問題具體分析但核心判斷邏輯是一樣的時間是數(shù)據(jù)的屬性還是系統(tǒng)的屬性決定了你用哪種語義。最后分享一個我自己的習(xí)慣任何涉及時間的 Flink SQL 作業(yè)上線前我都會在 DDL 旁邊寫一行注釋說明為什么選這種時間語義、水位線為什么設(shè)這個值。過幾個月回頭看這行注釋能省下大量重新推理的時間。時間語義這東西寫的時候覺得理所當(dāng)然改的時候才發(fā)現(xiàn)處處是坑。