開頭先聊點(diǎn)實(shí)際的。我做了幾年數(shù)倉(cāng)開發(fā)接觸Flink之前一直有個(gè)錯(cuò)覺覺得這是搞實(shí)時(shí)計(jì)算的團(tuán)隊(duì)才需要碰的東西離線數(shù)倉(cāng)每天跑批、出報(bào)表跟流式計(jì)算八竿子打不著。直到某公司內(nèi)部的實(shí)時(shí)數(shù)倉(cāng)項(xiàng)目落地我才發(fā)現(xiàn)自己對(duì)Flink的認(rèn)知完全不夠用——離線數(shù)倉(cāng)那套“T1”的思路在實(shí)時(shí)場(chǎng)景里根本行不通而Flink恰恰補(bǔ)上了這塊短板?，F(xiàn)在回頭看數(shù)倉(cāng)開發(fā)工程師不懂Flink概念就像寫后端不懂緩存一樣遲早會(huì)在業(yè)務(wù)面前吃癟。這篇文章我會(huì)圍繞數(shù)倉(cāng)開發(fā)中Flink的核心概念展開把時(shí)間語義、Watermark、窗口、狀態(tài)、檢查點(diǎn)這些繞不開的東西講清楚也會(huì)聊聊實(shí)時(shí)數(shù)倉(cāng)分層設(shè)計(jì)、Flink SQL實(shí)操、維表關(guān)聯(lián)、CDC入倉(cāng)這些落地時(shí)要面對(duì)的問題。內(nèi)容不追求面面俱到但保證是實(shí)際開發(fā)中真正常用的那部分。適合兩類人看一是剛接觸實(shí)時(shí)數(shù)倉(cāng)、想弄明白Flink到底解決了什么問題的新手二是在離線數(shù)倉(cāng)待了幾年、想轉(zhuǎn)實(shí)時(shí)方向但被一堆概念勸退的開發(fā)。1. 先搞清楚Flink在數(shù)倉(cāng)開發(fā)里到底扮演什么角色1.1 數(shù)倉(cāng)開發(fā)口中的“Flink”通常指什么很多做數(shù)倉(cāng)的同學(xué)第一次接觸Flink是從“實(shí)時(shí)計(jì)算”這四個(gè)字開始的。但Flink本身不是一個(gè)黑盒的實(shí)時(shí)計(jì)算引擎它是一個(gè)分布式處理引擎既能處理無界流數(shù)據(jù)也能處理有界批數(shù)據(jù)。所謂無界流就是數(shù)據(jù)一直在產(chǎn)生、永遠(yuǎn)沒有終點(diǎn)比如用戶點(diǎn)擊日志、訂單流水、交易明細(xì)有界數(shù)據(jù)就是我們熟悉的離線批次比如前一天的訂單表、用戶表。Flink在兩套場(chǎng)景下都能跑這也是它和其他計(jì)算引擎最大的差異點(diǎn)。早期我們做數(shù)倉(cāng)離線用Hive跑SQL實(shí)時(shí)用Spark Streaming或者Storm兩套引擎、兩套代碼、兩套運(yùn)維邏輯。Flink出現(xiàn)之后一套引擎可以同時(shí)處理批和流代碼層面還能做到流批共享同一套邏輯這就是常說的“流批一體”。對(duì)數(shù)倉(cāng)開發(fā)來說這意味著不用再維護(hù)兩套計(jì)算體系學(xué)習(xí)成本、維護(hù)成本都在降。但實(shí)際落地時(shí)要清醒一點(diǎn)流批一體不是說你寫一份代碼就萬事大吉。批處理可以基于完整數(shù)據(jù)集做全局計(jì)算流處理面對(duì)的是源源不斷的數(shù)據(jù)天然要處理亂序、遲到、窗口計(jì)算這些問題。Flink只是提供了統(tǒng)一的編程模型和底層能力具體怎么設(shè)計(jì)作業(yè)還是要靠開發(fā)自己對(duì)業(yè)務(wù)的理解。1.2 數(shù)倉(cāng)里Flink的典型用武之地?cái)?shù)倉(cāng)開發(fā)接觸Flink最常見的場(chǎng)景集中在下面幾個(gè)方向。第一個(gè)是實(shí)時(shí)ETL。離線ETL是把數(shù)據(jù)從源端抽到數(shù)倉(cāng)清洗、轉(zhuǎn)換、加載T1產(chǎn)出。實(shí)時(shí)ETL就是把這個(gè)過程縮短到秒級(jí)甚至毫秒級(jí)數(shù)據(jù)從業(yè)務(wù)庫或日志系統(tǒng)產(chǎn)生后經(jīng)過Flink作業(yè)清洗過濾直接寫入下游的Kafka主題或者實(shí)時(shí)數(shù)倉(cāng)存儲(chǔ)。比如訂單數(shù)據(jù)產(chǎn)生后要從JSON串里抽取關(guān)鍵字段、補(bǔ)上維度信息、過濾掉無效數(shù)據(jù)再轉(zhuǎn)發(fā)給下游使用方。第二個(gè)是維表關(guān)聯(lián)。實(shí)時(shí)業(yè)務(wù)場(chǎng)景里訂單流通常只包含訂單ID、商品ID、用戶ID這類主鍵但下游指標(biāo)計(jì)算需要商品名稱、類目、用戶等級(jí)這些維度信息。Flink SQL里可以通過維表關(guān)聯(lián)把維度字段補(bǔ)全到主流上這是離線數(shù)倉(cāng)里JOIN維表的實(shí)時(shí)版本。第三個(gè)是實(shí)時(shí)指標(biāo)計(jì)算。GMV、DAU、下單轉(zhuǎn)化率、實(shí)時(shí)庫存這些指標(biāo)如果等離線任務(wù)算完再出業(yè)務(wù)基本沒法用。Flink可以基于事件時(shí)間做窗口聚合每五分鐘、每十分鐘輸出一次實(shí)時(shí)結(jié)果業(yè)務(wù)側(cè)拿到的數(shù)據(jù)接近實(shí)時(shí)。第四個(gè)是CDC數(shù)據(jù)入倉(cāng)。業(yè)務(wù)數(shù)據(jù)庫的變更數(shù)據(jù)通過CDC工具捕獲發(fā)送到KafkaFlink消費(fèi)后解析成統(tǒng)一格式寫入數(shù)據(jù)湖或數(shù)倉(cāng)存儲(chǔ)。這也是近幾年實(shí)時(shí)數(shù)倉(cāng)建設(shè)里很熱的鏈路后面細(xì)說。1.3 流批一體為什么成了必選項(xiàng)流批一體這個(gè)概念在前幾年還是個(gè)趨勢(shì)現(xiàn)在已經(jīng)成了數(shù)倉(cāng)建設(shè)的基本要求。原因很樸素業(yè)務(wù)不希望同一套邏輯在實(shí)時(shí)和離線算出來結(jié)果不一致。我踩過的坑就是典型例子。某次做用戶留存指標(biāo)離線鏈路用Hive SQL口徑統(tǒng)計(jì)實(shí)時(shí)鏈路用Flink SQL實(shí)現(xiàn)兩邊代碼是不同人寫的join邏輯、去重方式有細(xì)微差異結(jié)果T1的數(shù)和對(duì)不上實(shí)時(shí)趨勢(shì)數(shù)業(yè)務(wù)方追著問哪個(gè)準(zhǔn)。最后只能靠人工校準(zhǔn)效率很低。流批一體的價(jià)值就在這里Flink SQL的語法體系同時(shí)面向批和流同樣的SQL邏輯跑在批模式產(chǎn)出的結(jié)果和跑在流模式產(chǎn)出的結(jié)果在語義上保持一致。你寫一遍邏輯既可以在凌晨跑批任務(wù)補(bǔ)全歷史數(shù)據(jù)也可以在白天跑流任務(wù)處理實(shí)時(shí)數(shù)據(jù)省去了兩套口徑對(duì)齊的痛苦。當(dāng)然說“保持一致”是理論層面實(shí)際使用中還是有一些細(xì)節(jié)差異需要注意后面講時(shí)間語義和狀態(tài)時(shí)會(huì)提到。2. 數(shù)倉(cāng)開發(fā)必須掌握的核心概念2.1 時(shí)間語義處理時(shí)間還是事件時(shí)間Flink里最基礎(chǔ)也最容易讓數(shù)倉(cāng)開發(fā)暈掉的概念就是時(shí)間語義。它分為兩種Processing Time處理時(shí)間和 Event Time事件時(shí)間。處理時(shí)間指的是數(shù)據(jù)到達(dá)Flink作業(yè)時(shí)處理算子所在機(jī)器的系統(tǒng)時(shí)間。這個(gè)時(shí)間跟數(shù)據(jù)本身沒關(guān)系數(shù)據(jù)早到晚到都會(huì)取當(dāng)前系統(tǒng)時(shí)間。事件時(shí)間指的是數(shù)據(jù)產(chǎn)生的時(shí)間通常由業(yè)務(wù)系統(tǒng)寫入數(shù)據(jù)中的某個(gè)字段。比如訂單日志里記錄的“下單時(shí)間”用戶點(diǎn)擊行為里記錄的“點(diǎn)擊時(shí)間”。離線數(shù)倉(cāng)里我們習(xí)慣用業(yè)務(wù)時(shí)間做統(tǒng)計(jì)比如算某天的GMV口徑肯定是“下單時(shí)間屬于那天”的數(shù)據(jù)。實(shí)時(shí)場(chǎng)景如果直接用處理時(shí)間問題就大了——數(shù)據(jù)在網(wǎng)絡(luò)傳輸、Kafka緩沖、算子處理過程中都會(huì)產(chǎn)生延遲凌晨下單的數(shù)據(jù)可能早上九點(diǎn)才到達(dá)Flink作業(yè)你要是按處理時(shí)間統(tǒng)計(jì)這筆訂單就算到了早上的窗口里結(jié)果就錯(cuò)了。所以實(shí)時(shí)數(shù)倉(cāng)開發(fā)里口徑正確的核心是事件時(shí)間。Flink作業(yè)要從數(shù)據(jù)里抽出事件時(shí)間字段用它驅(qū)動(dòng)窗口計(jì)算。這也是為什么Flink要求數(shù)據(jù)源里必須有一個(gè)能代表業(yè)務(wù)時(shí)間的字段沒有的話只能退而求其次用處理時(shí)間但結(jié)果精度就要打折。2.2 Watermark與亂序數(shù)據(jù)處理說事件時(shí)間就繞不開Watermark。簡(jiǎn)單理解Watermark是Flink用來判斷“數(shù)據(jù)遲到到什么程度就不再等了”的機(jī)制。流式數(shù)據(jù)天生亂序業(yè)務(wù)系統(tǒng)產(chǎn)生數(shù)據(jù)的時(shí)間有先后但到達(dá)Flink作業(yè)的時(shí)間可能是亂的。比如訂單A在10:00:01產(chǎn)生訂單B在10:00:02產(chǎn)生但網(wǎng)絡(luò)抖動(dòng)導(dǎo)致B先到、A后到。如果你的窗口是10:00:00到10:00:10Flink不可能永遠(yuǎn)等下去因?yàn)閿?shù)據(jù)流是無界的你不往下走結(jié)果永遠(yuǎn)出不來。Watermark機(jī)制解決的就是這個(gè)問題。作業(yè)會(huì)維護(hù)一個(gè)Watermark值表示“事件時(shí)間早于這個(gè)值的數(shù)據(jù)都已經(jīng)到達(dá)了”。比如Watermark推進(jìn)到10:00:08就認(rèn)為10:00:08之前的亂序數(shù)據(jù)都來了可以關(guān)窗計(jì)算。那10:00:08之后又來的遲到數(shù)據(jù)怎么辦可以配置allowedLateness給窗口留一段等待時(shí)間還可以把遲到數(shù)據(jù)單獨(dú)輸出到側(cè)輸出流里處理。實(shí)際開發(fā)里設(shè)定Watermark要平衡兩個(gè)訴求等太久會(huì)拖慢實(shí)時(shí)性等太短會(huì)導(dǎo)致數(shù)據(jù)遺漏。我的經(jīng)驗(yàn)是先摸清數(shù)據(jù)源在正常情況下的延遲分布看95分位或者99分位的延遲是多少Watermark偏移量可以設(shè)置為比這個(gè)稍大一點(diǎn)。如果業(yè)務(wù)對(duì)精確度要求高可以適當(dāng)多等但也別過頭否則實(shí)時(shí)指標(biāo)的“實(shí)時(shí)”就名存實(shí)亡了。2.3 窗口實(shí)時(shí)數(shù)倉(cāng)的“批次”邏輯離線數(shù)倉(cāng)天生有批次概念數(shù)據(jù)按天分區(qū)跑批就是處理一個(gè)分區(qū)的數(shù)據(jù)。流式計(jì)算沒有自然邊界數(shù)據(jù)一直在跑必須有機(jī)制把無界流切成一段一段有界的集合這就是窗口。Flink里常用的是滾動(dòng)窗口Tumbling Window、滑動(dòng)窗口Sliding Window、會(huì)話窗口Session Window。滾動(dòng)窗口是把數(shù)據(jù)按固定時(shí)間長(zhǎng)度切分窗口之間互不重疊比如每5分鐘一個(gè)窗口第一個(gè)窗口收集10:00到10:05的數(shù)據(jù)第二個(gè)窗口收集10:05到10:10的數(shù)據(jù)。滑動(dòng)窗口是每隔一段時(shí)間滑動(dòng)一次窗口長(zhǎng)度可以大于滑動(dòng)步長(zhǎng)數(shù)據(jù)會(huì)在多個(gè)窗口里被重復(fù)計(jì)算適合看“過去10分鐘的累計(jì)值每隔1分鐘更新一次”這類需求。會(huì)話窗口根據(jù)數(shù)據(jù)活躍間隙切分如果數(shù)據(jù)停了超過指定時(shí)間就結(jié)束當(dāng)前窗口適合分析用戶訪問會(huì)話。數(shù)倉(cāng)開發(fā)剛上手時(shí)最容易踩的坑是把窗口當(dāng)成離線SQL里的GROUP BY。離線分組是把完整數(shù)據(jù)集按維度聚合窗口只是把無界流切塊再在塊內(nèi)做聚合。窗口的劃分是基于事件時(shí)間還是處理時(shí)間直接決定結(jié)果是否準(zhǔn)確這又回到了前面講的時(shí)間語義。另外窗口內(nèi)做聚合要留意狀態(tài)的獲取方式。Flink SQL的窗口聚合語法比較簡(jiǎn)潔比如用TUMBLE(time_attr, INTERVAL 5 MINUTE)定義五分鐘滾動(dòng)窗口但底層會(huì)為每個(gè)窗口保存狀態(tài)窗口數(shù)量太多時(shí)狀態(tài)壓力會(huì)很大。如果業(yè)務(wù)允許盡量用較大的窗口粒度減少狀態(tài)量。2.4 狀態(tài)與狀態(tài)后端Flink的“內(nèi)存賬本”狀態(tài)這個(gè)概念離線數(shù)倉(cāng)開發(fā)一開始不太容易理解因?yàn)殡x線任務(wù)是無狀態(tài)的——Hive跑SQL輸入一批數(shù)據(jù)輸出一批結(jié)果任務(wù)結(jié)束就沒有然后了。Flink作業(yè)不同它是長(zhǎng)期運(yùn)行的同一個(gè)作業(yè)要處理連續(xù)不斷的數(shù)據(jù)流算子需要記住一些信息這些信息就是狀態(tài)。舉個(gè)例子實(shí)時(shí)計(jì)算用戶的累計(jì)消費(fèi)金額你要記住每個(gè)用戶當(dāng)前已經(jīng)消費(fèi)了多少每來一條新訂單就在原值上加一下這個(gè)“當(dāng)前累計(jì)值”就是狀態(tài)。再比如去重統(tǒng)計(jì)每日活躍用戶你要記住今天已經(jīng)看到哪些用戶了才能決定新來的用戶要不要計(jì)入DAU這個(gè)“已見用戶集合”也是狀態(tài)。數(shù)倉(cāng)開發(fā)做實(shí)時(shí)計(jì)算一定要有狀態(tài)管理的意識(shí)因?yàn)闋顟B(tài)的大小直接決定作業(yè)能不能長(zhǎng)期穩(wěn)定運(yùn)行。默認(rèn)情況下狀態(tài)存在內(nèi)存里訪問快但容量有限狀態(tài)太大會(huì)導(dǎo)致內(nèi)存溢出這時(shí)可以換成RocksDB狀態(tài)后端把狀態(tài)落到磁盤容量大得多但讀寫速度略慢。實(shí)際項(xiàng)目里幾GB以下的內(nèi)存狀態(tài)還行往上走基本都要切RocksDB。狀態(tài)還需要定期清理。無限增長(zhǎng)的KeyedState如果一直不清理作業(yè)會(huì)越來越慢最終OOM。解決辦法是給狀態(tài)設(shè)置TTL比如用戶維表狀態(tài)保留24小時(shí)超過時(shí)間的key自動(dòng)過期這樣狀態(tài)體積能控制在一定范圍內(nèi)。2.5 檢查點(diǎn)與精確一次語義提到Flink大家還會(huì)經(jīng)常聽到“精確一次語義Exactly-Once”和“檢查點(diǎn)Checkpoint”。理解這兩個(gè)概念對(duì)定位數(shù)據(jù)重復(fù)或丟失問題很有幫助。先通俗解釋精確一次。上游數(shù)據(jù)源在故障恢復(fù)時(shí)可能會(huì)重新發(fā)送已發(fā)送過的數(shù)據(jù)如果Flink作業(yè)不處理這種重復(fù)那聚合結(jié)果就會(huì)多算一條。Flink通過“檢查點(diǎn)機(jī)制 端到端的事務(wù)性寫入”盡可能保證每條數(shù)據(jù)只對(duì)結(jié)果產(chǎn)生一次影響這就是精確一次。注意這里我說“盡可能”因?yàn)槎说蕉说木_一次還依賴下游存儲(chǔ)和上游數(shù)據(jù)源的配合比如Kafka支持事務(wù)、數(shù)據(jù)源支持回放鏈路里任何一環(huán)不支持整體就只能退化為At-Least-Once至少一次結(jié)果就可能出現(xiàn)重復(fù)。檢查點(diǎn)是Flink用來實(shí)現(xiàn)數(shù)據(jù)一致性的核心機(jī)制。作業(yè)會(huì)周期性生成一個(gè)快照記錄當(dāng)前處理到的數(shù)據(jù)位置和各算子狀態(tài)類似離線數(shù)倉(cāng)的“數(shù)據(jù)備份”。當(dāng)作業(yè)發(fā)生故障重啟時(shí)可以從最近的檢查點(diǎn)恢復(fù)之前處理完的數(shù)據(jù)不用重算這個(gè)恢復(fù)機(jī)制叫“從檢查點(diǎn)恢復(fù)”。我用通俗的類比講一下檢查點(diǎn)相當(dāng)于游戲存檔打到一半掛了不用從頭開始直接讀檔繼續(xù)。沒有檢查點(diǎn)的話Flink作業(yè)掛掉重啟要么丟數(shù)據(jù)、要么全量重算在實(shí)時(shí)場(chǎng)景里都是不可接受的。實(shí)際配置時(shí)要注意檢查點(diǎn)生成頻率和間隔太頻繁會(huì)占用IO影響性能太稀疏會(huì)導(dǎo)致恢復(fù)時(shí)重放大量數(shù)據(jù)。常用策略是設(shè)置1到3分鐘一個(gè)檢查點(diǎn)同時(shí)開啟“非對(duì)齊檢查點(diǎn)”選項(xiàng)處理背壓場(chǎng)景這些配置在項(xiàng)目里跑了很久驗(yàn)證下來是比較穩(wěn)的。3. 落地到數(shù)倉(cāng)業(yè)務(wù)的實(shí)操要點(diǎn)3.1 實(shí)時(shí)數(shù)倉(cāng)分層怎么設(shè)計(jì)離線數(shù)倉(cāng)有ODS、DWD、DWS、ADS分層實(shí)時(shí)數(shù)倉(cāng)也沿用了類似思路但實(shí)現(xiàn)方式和細(xì)節(jié)差異明顯。實(shí)時(shí)數(shù)倉(cāng)的ODS層通常是Kafka里的原始主題。業(yè)務(wù)日志經(jīng)采集工具進(jìn)入KafkaFlink作業(yè)直接消費(fèi)這些原始數(shù)據(jù)做最基礎(chǔ)的格式解析輸出到Kafka的ODS主題。注意ODS層不建議放太多業(yè)務(wù)邏輯保持原始方便后續(xù)回溯和重新加工。DWD層做清洗和維度補(bǔ)充。比如訂單數(shù)據(jù)要過濾掉測(cè)試訂單、退款訂單要解析JSON字段、把時(shí)間字段標(biāo)準(zhǔn)化還可以通過維表關(guān)聯(lián)補(bǔ)上商品和用戶維度信息處理完的數(shù)據(jù)落到Kafka的DWD主題。這一層是實(shí)時(shí)數(shù)倉(cāng)里工作量最大的環(huán)節(jié)大量Flink SQL作業(yè)集中在這一層。DWS層做匯總。按業(yè)務(wù)需要的維度做輕度聚合比如每分鐘的訂單數(shù)、銷售額、活躍用戶數(shù)產(chǎn)出到Kafka或者StarRocks、Doris這類OLAP存儲(chǔ)。ADS層直接對(duì)接業(yè)務(wù)應(yīng)用比如實(shí)時(shí)大屏、實(shí)時(shí)報(bào)表、實(shí)時(shí)告警數(shù)據(jù)從DWS層查詢或進(jìn)一步加工。這套分層設(shè)計(jì)最大的好處是復(fù)用。DWD層的數(shù)據(jù)經(jīng)過清洗補(bǔ)全后下游多個(gè)應(yīng)用可以共享同一份高質(zhì)量數(shù)據(jù)避免每個(gè)應(yīng)用各寫一套解析邏輯導(dǎo)致口徑不一致。這一點(diǎn)和離線數(shù)倉(cāng)的“統(tǒng)一出口”思路是一致的。3.2 Flink SQL開發(fā)的核心實(shí)操現(xiàn)在實(shí)時(shí)數(shù)倉(cāng)開發(fā)的主力語言是Flink SQL比DataStream API上手門檻低很多對(duì)從Hive SQL轉(zhuǎn)過來的數(shù)倉(cāng)開發(fā)尤其友好。一份典型的Flink SQL作業(yè)邏輯通常分三步建源表、建目標(biāo)表、寫插入語句。建源表要指定連接器類型、Kafka地址、topic、數(shù)據(jù)格式、事件時(shí)間字段和Watermark策略。比如消費(fèi)訂單數(shù)據(jù)可以在CREATE TABLE語句里定義字段然后根據(jù)訂單時(shí)間字段聲明Watermark延遲時(shí)間設(shè)30秒。建目標(biāo)表則是定義結(jié)果要寫入的存儲(chǔ)比如Kafka目標(biāo)主題或者StarRocks表。寫插入語句時(shí)可以像寫Hive一樣寫SELECT但要注意Flink SQL流式處理的特性不是每個(gè)子查詢都能在流上跑比如沒有窗口的無限流上做復(fù)雜相關(guān)子查詢就可能產(chǎn)生“查詢沒有在有限窗口內(nèi)執(zhí)行”的報(bào)錯(cuò)。遇到這類問題思路要轉(zhuǎn)換回來——先想想這個(gè)計(jì)算在流上有沒有邊界窗口能不能切出邊界來。動(dòng)態(tài)表的理解也很有用。Flink SQL里表中的數(shù)據(jù)是實(shí)時(shí)變化的每次查詢相當(dāng)于對(duì)動(dòng)態(tài)表的快照做計(jì)算結(jié)果也是動(dòng)態(tài)表。離線Hive的靜態(tài)表思維要調(diào)整過來不能再假設(shè)表內(nèi)容固定不變。3.3 維表關(guān)聯(lián)的常用方式實(shí)時(shí)數(shù)倉(cāng)開發(fā)里維表關(guān)聯(lián)是個(gè)繞不開的大話題。離線Hive里做維度補(bǔ)全很簡(jiǎn)單一個(gè)JOIN搞定數(shù)據(jù)全量在那隨便關(guān)聯(lián)。實(shí)時(shí)場(chǎng)景不一樣主流數(shù)據(jù)是無限的流維度數(shù)據(jù)在外部存儲(chǔ)里比如MySQL、Redis、HBase不能直接JOIN得想辦法查詢。Flink SQL里最常用的維表關(guān)聯(lián)方式是Lookup Join。Lookup Join的原理是當(dāng)主流數(shù)據(jù)流經(jīng)過算子時(shí)根據(jù)關(guān)聯(lián)字段實(shí)時(shí)去查詢外部維表把查詢到的維度字段補(bǔ)到當(dāng)前數(shù)據(jù)上。可以是每來一條數(shù)據(jù)查一次維表也可以配置緩存把查過的維度存起來一段時(shí)間內(nèi)復(fù)用減少外部存儲(chǔ)壓力。這里幾個(gè)參數(shù)很關(guān)鍵緩存大小、緩存TTL、最大重試次數(shù)。緩存大小決定內(nèi)存里能放多少維度數(shù)據(jù)緩存TTL決定維度數(shù)據(jù)的有效時(shí)長(zhǎng)最大重試次數(shù)是維表查詢失敗時(shí)最多重試幾次。實(shí)際項(xiàng)目里維表通常放Redis或者M(jìn)ySQL我建議優(yōu)先用Redis查詢快、壓力小數(shù)據(jù)量小的維表也可以內(nèi)存緩存效率更高。另一個(gè)點(diǎn)是不管用哪種維表都要考慮到維度數(shù)據(jù)更新的問題。比如商品類目調(diào)整了緩存里還是舊值下游指標(biāo)就會(huì)用到舊維度。解決辦法是把緩存TTL控制在可接受范圍內(nèi)或者通過CDC把維度變更也實(shí)時(shí)同步到緩存中。3.4 CDC數(shù)據(jù)入倉(cāng)鏈路解析CDCChange Data Capture變更數(shù)據(jù)捕獲是近幾年實(shí)時(shí)數(shù)倉(cāng)最熱的方向核心是把業(yè)務(wù)數(shù)據(jù)庫的變更記錄增刪改實(shí)時(shí)捕獲出來同步到數(shù)倉(cāng)或數(shù)據(jù)湖。鏈路通常是這樣的業(yè)務(wù)MySQL產(chǎn)生變更日志BinlogCDC工具把日志解析成統(tǒng)一的變更事件發(fā)送到KafkaFlink消費(fèi)后寫入數(shù)倉(cāng)存儲(chǔ)。很多數(shù)倉(cāng)開發(fā)對(duì)CDC的理解是“數(shù)據(jù)庫同步工具”而已容易忽略它在實(shí)時(shí)數(shù)倉(cāng)里的真正意義。有了CDC業(yè)務(wù)表的數(shù)據(jù)可以實(shí)時(shí)進(jìn)入數(shù)倉(cāng)ODS層不再是T1全量快照而是按變更時(shí)間形成的實(shí)時(shí)流水下游可以隨時(shí)基于最新數(shù)據(jù)計(jì)算同時(shí)保留完整變更歷史支持回溯。Flink在CDC鏈路里承擔(dān)的任務(wù)通常是數(shù)據(jù)解析和格式轉(zhuǎn)換。業(yè)務(wù)庫的Binlog格式是MySQL相關(guān)的Flink CDC連接器會(huì)解析成統(tǒng)一的RowDataFlink作業(yè)再接入自己的邏輯。也常見用Flink做全量加增量同步先一次性讀取存量數(shù)據(jù)再接續(xù)讀取變更數(shù)據(jù)保證上游表數(shù)據(jù)完整。這個(gè)過程要注意主鍵去重和schema演進(jìn)的兼容否則同步到一半表結(jié)構(gòu)變了作業(yè)會(huì)掛。寫入目標(biāo)存儲(chǔ)時(shí)建議優(yōu)先選支持upsert的存儲(chǔ)比如Hudi、Iceberg、Kafka。傳統(tǒng)Hive分區(qū)表不適合高頻upsert因?yàn)镠ive本身不支持行級(jí)更新CDC數(shù)據(jù)寫進(jìn)去會(huì)產(chǎn)生大量小文件。這個(gè)坑我踩過后期靠Hudi的mor表才把問題解決。4. 參數(shù)調(diào)優(yōu)與問題排查實(shí)錄4.1 作業(yè)并行度與資源配置離線任務(wù)跑掛了重跑就行實(shí)時(shí)作業(yè)不一樣它7x24小時(shí)運(yùn)行資源分配不對(duì)問題會(huì)積累最終拖垮整個(gè)鏈路。Flink作業(yè)的并行度是第一個(gè)要調(diào)的參數(shù)它決定了作業(yè)的算子有多少個(gè)并行實(shí)例每個(gè)實(shí)例處理一部分?jǐn)?shù)據(jù)。并行度不是越大越好。我見過有人無腦把并行度調(diào)到64結(jié)果Kafka分區(qū)數(shù)只有12導(dǎo)致大量并行子任務(wù)空閑白白占用資源。合理做法是先確認(rèn)上游Kafka分區(qū)數(shù)并行度一般不要超過分區(qū)數(shù)否則單個(gè)分區(qū)的數(shù)據(jù)被拆到多個(gè)并行實(shí)例里狀態(tài)一致性和順序性都可能出問題。內(nèi)存參數(shù)也很關(guān)鍵。TaskManager的總內(nèi)存、托管內(nèi)存、JVM堆外內(nèi)存這幾個(gè)參數(shù)要區(qū)分清楚。狀態(tài)大小直接決定托管內(nèi)存的配置如果作業(yè)用了RocksDB狀態(tài)后端托管內(nèi)存要留足否則狀態(tài)頻繁落盤會(huì)拖慢性能。實(shí)際調(diào)優(yōu)時(shí)先看監(jiān)控面板里內(nèi)存使用率確認(rèn)瓶頸在哪再針對(duì)性調(diào)整參數(shù)不要一次性把所有參數(shù)都改了否則出了問題沒法定位是哪個(gè)參數(shù)引起的。4.2 反壓?jiǎn)栴}怎么定位反壓Backpressure是實(shí)時(shí)計(jì)算里最讓人頭疼的問題之一。通俗解釋數(shù)據(jù)從上游算子流到下游算子下游處理速度跟不上上游生產(chǎn)速度數(shù)據(jù)會(huì)在鏈路中堆積就像水管下游堵住了水壓會(huì)往上傳遞。反壓出現(xiàn)時(shí)作業(yè)不會(huì)直接報(bào)錯(cuò)退出但數(shù)據(jù)延遲會(huì)越來越大checkpoint耗時(shí)會(huì)變長(zhǎng)甚至超時(shí)失敗最終導(dǎo)致整個(gè)作業(yè)無法正常工作。定位反壓我習(xí)慣先在Flink Web UI上看每個(gè)算子的“背壓狀態(tài)”從處理速度最慢的算子入手確認(rèn)是計(jì)算邏輯復(fù)雜、調(diào)用外部服務(wù)慢還是Sink寫入性能差。如果反壓發(fā)生在Sink階段優(yōu)先檢查下游存儲(chǔ)是否夠快。比如寫入StarRocks時(shí)如果攢批策略不合理每條數(shù)據(jù)都觸發(fā)寫入吞吐肯定上不去。這時(shí)要調(diào)整攢批大小、刷新間隔、重試次數(shù)這些參數(shù)。如果反壓發(fā)生在Join操作多半是維表Lookup慢先看維表緩存命中率命中率低就擴(kuò)大緩存或加Redis集群。我的習(xí)慣是“反壓?jiǎn)栴}先查下游再查中間算子”因?yàn)閷?shí)時(shí)鏈路里Sink往往是全局瓶頸。反壓只是現(xiàn)象真正要解決的是瓶頸算子的處理能力問題。4.3 狀態(tài)過大與數(shù)據(jù)傾斜長(zhǎng)時(shí)間運(yùn)行的Flink作業(yè)狀態(tài)膨脹和數(shù)據(jù)傾斜是兩個(gè)會(huì)慢慢浮現(xiàn)的殺手。狀態(tài)過大的典型特征是checkpoint越來越慢恢復(fù)時(shí)間越來越久最終作業(yè)重啟一次要好幾分鐘。解決辦法除了前面說的清理邏輯、設(shè)置TTL還可以考慮兩點(diǎn)一是檢查是否有不必要的key維度把狀態(tài)粒度打細(xì)了比如按訂單ID作為key會(huì)產(chǎn)生海量小狀態(tài)二是優(yōu)化聚合邏輯盡量分層聚合先局部聚合再全局聚合減少狀態(tài)量。數(shù)據(jù)傾斜在Flink里和離線一樣常見。比如實(shí)時(shí)計(jì)算某類商品的銷售指標(biāo)爆款商品的key數(shù)據(jù)量可能是其他商品的幾百倍單個(gè)并行子任務(wù)會(huì)被熱點(diǎn)數(shù)據(jù)壓垮其他子任務(wù)空轉(zhuǎn)。離線可以加隨機(jī)前綴打散Flink里思路類似在key上拼隨機(jī)數(shù)做局部聚合再按真實(shí)key做最終聚合。優(yōu)先級(jí)高的傾斜場(chǎng)景比如大主播直播期間的訂單流不提前處理作業(yè)基本撐不過大促峰值。傾斜問題的排查要靠監(jiān)控各個(gè)子任務(wù)的處理記錄數(shù)和延遲發(fā)現(xiàn)某個(gè)子任務(wù)數(shù)據(jù)量明顯高于均值基本就是傾斜了。Flink Web UI的Subtask指標(biāo)比離線任務(wù)的日志直觀得多開發(fā)一定要養(yǎng)成看指標(biāo)面板的習(xí)慣。4.4 常見報(bào)錯(cuò)速查表實(shí)時(shí)作業(yè)跑起來之后報(bào)錯(cuò)信息千奇百怪但排在最前面的高頻問題就那么幾類整理成表格方便對(duì)照。報(bào)錯(cuò)場(chǎng)景常見原因排查思路作業(yè)啟動(dòng)后很快失敗提示序列化異常POJO類型缺少無參構(gòu)造或字段未注冊(cè)檢查Flink SQL定義的數(shù)據(jù)類型與源數(shù)據(jù)是否一致checkpoint失敗持續(xù)反壓下游處理慢或狀態(tài)過大先定位反壓算子再調(diào)整Sink寫參數(shù)或清理狀態(tài)維表關(guān)聯(lián)查詢超時(shí)外部存儲(chǔ)壓力大或緩存命中率低查看維表查詢耗時(shí)加大緩存或優(yōu)化查詢邏輯數(shù)據(jù)重復(fù)嚴(yán)重上游At-Least-Once導(dǎo)致重復(fù)消費(fèi)檢查是否開啟Checkpoint下游是否支持冪等寫入窗口結(jié)果一直不出Watermark不推進(jìn)確認(rèn)事件時(shí)間字段是否配置正確Watermark延遲是否過大內(nèi)存溢出OOM狀態(tài)太大或并行度過高檢查托管內(nèi)存配置切換RocksDB狀態(tài)后端這些報(bào)錯(cuò)都有一個(gè)共同點(diǎn)日志只是表象真正的原因要看指標(biāo)和數(shù)據(jù)流特征。我處理這類問題時(shí)很少只看最后幾行報(bào)錯(cuò)而是先看作業(yè)拓?fù)湓倏疵總€(gè)算子的輸入輸出速率、Watermark延遲、checkpoint時(shí)長(zhǎng)把“現(xiàn)象-問題-優(yōu)化”串成一條線來排查。5. 參數(shù)調(diào)優(yōu)補(bǔ)充幾個(gè)容易被忽略的細(xì)節(jié)調(diào)優(yōu)這塊內(nèi)容單獨(dú)放一節(jié)講是因?yàn)楹芏囗?xiàng)目上線前期風(fēng)平浪靜跑了一兩周后才暴雷問題往往出在幾個(gè)容易被忽略的參數(shù)配置上。第一個(gè)是checkpoint的配置細(xì)節(jié)。很多團(tuán)隊(duì)只設(shè)置了檢查點(diǎn)間隔忽略了超時(shí)時(shí)間和失敗容忍次數(shù)。默認(rèn)超時(shí)時(shí)間如果偏大checkpoint失敗幾次后作業(yè)可能直接重啟導(dǎo)致長(zhǎng)時(shí)間的中斷。建議明確設(shè)置超時(shí)時(shí)間比如10分鐘同時(shí)設(shè)置checkpoint失敗容忍次數(shù)給短暫的外部抖動(dòng)留緩沖。第二個(gè)是空閑源Idle Source的處理。實(shí)時(shí)作業(yè)里如果有多個(gè)數(shù)據(jù)源其中某個(gè)數(shù)據(jù)源長(zhǎng)時(shí)間沒有新數(shù)據(jù)它的Watermark不推進(jìn)會(huì)導(dǎo)致整個(gè)作業(yè)的Watermark被拖住窗口一直不觸發(fā)。解決辦法是給源表設(shè)置idle超時(shí)參數(shù)讓長(zhǎng)時(shí)間空閑的數(shù)據(jù)源不再影響整體Watermark推進(jìn)。這個(gè)問題不遇到一次書面上很難有感覺但遇到后你會(huì)印象深刻。第三個(gè)是作業(yè)的并行子任務(wù)之間的數(shù)據(jù)傾斜監(jiān)控。Flink Web UI上可以看每個(gè)Subtask的“Records Sent”和“Records Received”如果發(fā)現(xiàn)某個(gè)Subtask接收的數(shù)據(jù)量是其他Subtask的幾十倍要第一時(shí)間處理。延遲處理傾斜問題的代價(jià)非常高因?yàn)闋顟B(tài)一旦被熱點(diǎn)key撐大后面想清理就難了。6. 流批一體的實(shí)踐邊界與選型思考寫到這里聊最后一個(gè)在數(shù)倉(cāng)開發(fā)中常見的問題流批一體到底能走到什么程度以及什么時(shí)候可以選擇不用Flink。理想狀態(tài)下一套Flink SQL邏輯既能跑批又能跑流實(shí)現(xiàn)“口徑統(tǒng)一、一套邏輯到處跑”。但實(shí)踐中我逐漸意識(shí)到流批一體更偏向“統(tǒng)一編程模型”和“統(tǒng)一SQL語法”并不是說同一個(gè)作業(yè)直接切換模式就能完全等價(jià)。批模式可以有界地掃描全量數(shù)據(jù)做一些依賴全局?jǐn)?shù)據(jù)的計(jì)算流模式要面對(duì)亂序、無界、增量計(jì)算有些SQL寫法在批上成立在流上會(huì)因?yàn)槿鄙俅翱谶吔缍鴪?bào)錯(cuò)。所以我的做法是核心指標(biāo)計(jì)算盡量用Flink SQL實(shí)現(xiàn)并且嚴(yán)格要求邏輯同時(shí)支持批和流兩種模式非核心的一次性分析任務(wù)仍交給Hive或者Spark跑沒必要強(qiáng)行統(tǒng)一。這樣既保證了主要指標(biāo)的實(shí)時(shí)離線一致性又不讓流批一體的理念成為所有任務(wù)的負(fù)擔(dān)。選型思考也簡(jiǎn)單說兩句。如果業(yè)務(wù)場(chǎng)景是分鐘級(jí)延遲就能滿足對(duì)數(shù)據(jù)準(zhǔn)確度要求極高那Kafka加Flink SQL加OLAP存儲(chǔ)的鏈路基本夠用沒必要上更重的架構(gòu)。如果需要處理非常復(fù)雜的實(shí)時(shí)計(jì)算邏輯比如多流關(guān)聯(lián)的復(fù)雜事件處理可以考慮DataStream API配合狀態(tài)編程靈活度更高但開發(fā)維護(hù)成本也更高。具體怎么選回到業(yè)務(wù)需求本身實(shí)時(shí)性要求、數(shù)據(jù)量級(jí)、團(tuán)隊(duì)熟悉度這三個(gè)因素基本能幫你定方向。最后再分享一個(gè)小技巧。我每次新建Flink作業(yè)都會(huì)先在測(cè)試環(huán)境用一份小的歷史數(shù)據(jù)回放一遍驗(yàn)證SQL邏輯和Watermark策略是否符合預(yù)期再切換到線上數(shù)據(jù)。別小看這個(gè)步驟實(shí)時(shí)作業(yè)一旦上線想再改邏輯就要承擔(dān)停機(jī)重放數(shù)據(jù)的成本遠(yuǎn)不如先在測(cè)試環(huán)境把坑踩完來得省事。做數(shù)倉(cāng)開發(fā)這幾年最大的體會(huì)是實(shí)時(shí)計(jì)算的門檻不在工具而在思維方式的轉(zhuǎn)變——從“離線全量”到“實(shí)時(shí)增量”從“靜態(tài)分批”到“動(dòng)態(tài)流式”想明白了這些Flink的概念再?gòu)?fù)雜也能慢慢變成趁手的工具。