據(jù)分析平臺實戰(zhàn)：從數(shù)據(jù)采集到可視化大屏)
1. 從需求到架構(gòu)先想清楚實時到底意味著什么接到一個實時數(shù)據(jù)分析平臺的需求時我心里第一反應(yīng)不是寫Flink代碼而是先問對方一句話你說的實時是秒級、分鐘級還是小時級這個問題問出來很多需求就瞬間清晰了。我見過太多團隊一上來就鋪Flink集群、搞大屏結(jié)果做了三個月發(fā)現(xiàn)核心指標(biāo)延遲五分鐘就能滿足白費了一堆功夫。Java工程師做實時數(shù)據(jù)平臺有個天然優(yōu)勢Flink本身就是Java/Scala生態(tài)你熟悉的Spring Boot、Maven、JVM調(diào)優(yōu)經(jīng)驗全部能復(fù)用。相比Python系或者純SQL系的數(shù)據(jù)棧Java團隊啃Flink的上手成本要低得多。這篇實戰(zhàn)文章我就圍繞數(shù)據(jù)采集→實時計算→可視化大屏這條完整鏈路把從零搭建一個實時數(shù)據(jù)分析平臺的架構(gòu)思路、核心代碼、踩坑記錄都講透。1.1 先拆需求你的大屏是不是假實時大多數(shù)實時數(shù)據(jù)分析平臺的真實需求拆開來看無非三類指標(biāo)監(jiān)控類比如訂單量、成交額、在線用戶數(shù)要求秒級或分鐘級刷新行為分析類用戶點擊流、頁面路徑要求準(zhǔn)實時但允許一定延遲預(yù)警通知類比如異常流量、交易失敗率飆升要求延遲越低越好這三類需求對技術(shù)選型的影響完全不同。我之前遇到一個做污水處理可視化大屏的項目客戶說實時結(jié)果詳細(xì)了解才知道污水?dāng)?shù)據(jù)本身是五分鐘采集一次那你就算用Flink做到毫秒級計算也沒有意義瓶頸在采集端。反過來如果是電商大促的實時成交大屏每秒鐘都有成千上萬條訂單事件那你就需要認(rèn)真設(shè)計從采集到展示的每一層。所以第一步永遠(yuǎn)是做延遲預(yù)算端到端延遲 采集延遲 傳輸延遲 計算延遲 存儲延遲 展示刷新延遲。把每一項都列出來標(biāo)出可接受范圍后續(xù)所有技術(shù)決策都有依據(jù)。1.2 端到端鏈路的分層設(shè)計我做的實時數(shù)據(jù)分析平臺標(biāo)準(zhǔn)鏈路分五層層級組件選型職責(zé)采集層Filebeat / Flink CDC / HTTP SDK將日志、數(shù)據(jù)庫變更、業(yè)務(wù)事件統(tǒng)一送入消息隊列傳輸層Kafka削峰填谷、緩沖削流解耦采集與計算計算層Flink實時ETL、窗口聚合、狀態(tài)計算、規(guī)則匹配存儲層Doris / ClickHouse / Redis結(jié)果表存儲、維度數(shù)據(jù)緩存、大屏查詢加速展示層Vue ECharts / DataV可視化大屏、指標(biāo)卡片、趨勢圖表這套鏈路跟傳統(tǒng)的離線數(shù)倉最大的區(qū)別在于數(shù)據(jù)不是按天批量加工而是以事件流的方式持續(xù)流動。Flink跑在Kafka和存儲之間相當(dāng)于一個永不停止的計算引擎——上游數(shù)據(jù)來了就算算完就寫寫完后端到端延遲通?？刂圃诿爰?。關(guān)于架構(gòu)理念現(xiàn)階段我做項目基本直接采用Kappa架構(gòu)思路不再搭建Lambda架構(gòu)。Lambda那套實時鏈路離線鏈路雙跑、最終結(jié)果合并的方案維護成本太高兩套代碼邏輯要一致本身就是災(zāi)難?，F(xiàn)在Flink的流批一體能力已經(jīng)相當(dāng)成熟一套代碼可以同時跑實時和離線Kappa架構(gòu)足夠覆蓋絕大多數(shù)場景。1.3 為什么選Flink而不是Spark Streaming每次做技術(shù)選型都要面對這個問題。我的答案很直接如果你的場景需要事件時間處理、精確一次語義、豐富的狀態(tài)管理Flink是當(dāng)前最優(yōu)解。事件時間處理數(shù)據(jù)在網(wǎng)絡(luò)上傳輸會有延遲和亂序Flink的Watermark機制可以基于事件真正發(fā)生的時間進行計算而不是基于數(shù)據(jù)到達時間。這在處理日志類數(shù)據(jù)時尤其重要——用戶點擊發(fā)生在10:00:00但因為網(wǎng)絡(luò)抖動這條日志10:00:10才到如果你用處理時間計算就把這10秒的誤差算進指標(biāo)里了。精確一次語義Exactly-OnceFlink通過Checkpoint 兩階段提交保證即使任務(wù)崩潰恢復(fù)數(shù)據(jù)也不會重復(fù)或丟失。做交易類指標(biāo)時這是剛需。狀態(tài)管理Flink可以把中間結(jié)果存在內(nèi)存或RocksDB中實現(xiàn)跨事件的聚合計算比如統(tǒng)計每個用戶的累計訪問次數(shù)這是純SQL流處理引擎很難做好的。當(dāng)然Spark Streaming在吞吐量上和微批處理也有自己的優(yōu)勢但說實話真心追求實時性的場景Flink的靈活性和生態(tài)完整度更適合。更何況現(xiàn)在Flink CDC已經(jīng)是數(shù)據(jù)庫實時采集的事實標(biāo)準(zhǔn)配合Java開發(fā)效率很高。2. 數(shù)據(jù)采集層的工程落地三種來源一套規(guī)范數(shù)據(jù)采集是整個實時鏈路的起點也是臟活累活最多的地方。很多同學(xué)把精力都花在Flink計算邏輯上結(jié)果數(shù)據(jù)源沒管好后面計算、展示全是垃圾進垃圾出。這里我按來源類型分開講。2.1 日志類采集Filebeat Kafka是黃金組合服務(wù)端日志是最常見的實時數(shù)據(jù)來源。我通常用Filebeat做日志采集器它比Flume輕量太多部署就是解壓一個二進制文件配置也簡單filebeat.inputs: - type: filestream enabled: true paths: - /data/logs/*.log fields: app_name: order-service log_type: business output.kafka: hosts: [kafka1:9092, kafka2:9092, kafka3:9092] topic: app-order-log partition.round_robin: reachable_only: true這個配置看起來簡單但有幾個細(xì)節(jié)務(wù)必注意不要用filestream直接用Kafka producer consumer方式Filebeat自帶背壓機制Kafka不可用時會暫停讀取本地文件不會丟數(shù)據(jù)。這是它作為采集端的核心理由。fields里打上應(yīng)用名和日志類型標(biāo)簽后面Flink消費時可以根據(jù)這些字段路由到不同處理邏輯。每個應(yīng)用單獨一個topic或者至少按業(yè)務(wù)線分topic。我曾經(jīng)見過所有應(yīng)用混在一個topic里的架構(gòu)Flink消費端要做大量過濾還會互相影響消費速度非常痛苦。2.2 數(shù)據(jù)庫變更采集Flink CDC到底怎么部署熱搜詞里flink cdc pipeline部署和flink cdc安裝部署出現(xiàn)頻率很高說明這個方向已經(jīng)成了實時數(shù)據(jù)平臺的主流需求。Flink CDC基于數(shù)據(jù)庫日志Binlog/Redo Log捕獲變更不打業(yè)務(wù)表對業(yè)務(wù)系統(tǒng)零侵入。部署上有兩種形態(tài)形態(tài)一Flink CDC作為Source接入Flink作業(yè)DataStreamSourceString stream env .addSource( MySqlSource.Stringbuilder() .hostname(localhost) .port(3306) .databaseList(shop) .tableList(shop.t_order) .username(cdc_user) .password(cdc_pwd) .deserializer(new JsonDebeziumDeserializationSchema()) .build() ) .setParallelism(1);這種形態(tài)適合在Flink作業(yè)里實時消費數(shù)據(jù)庫變更。注意setParallelism(1)很關(guān)鍵因為單個MySQL實例的Binlog讀取是單線程的并行度設(shè)置高了反而會出問題。形態(tài)二Flink CDC Pipeline獨立部署如果你的目標(biāo)是數(shù)據(jù)庫實時同步到另一個存儲可以用Flink CDC Pipeline也就是之前的CDAS它基于Yaml配置就能完成整庫同步不需要寫一行Java代碼source: type: mysql hostname: localhost port: 3306 username: cdc_user password: cdc_pwd tables: shop\.* sink: type: doris fenodes: doris:8030 username: admin password: admin123Pipeline形態(tài)適合快速落地但是如果你想在同步過程中做數(shù)據(jù)加工比如字段映射、類型轉(zhuǎn)換、過濾還是寫Java代碼更靈活。我的建議是同步裸數(shù)據(jù)用Pipeline需要加工用源碼。關(guān)于Flink CDC最大的坑是存量數(shù)據(jù)與增量數(shù)據(jù)的一致性問題。Flink CDC默認(rèn)會先做一次全量快照再切換到Binlog增量這個過程對數(shù)據(jù)庫有一定壓力。建議在業(yè)務(wù)低峰期做首次同步并且監(jiān)控好源庫的IOPS和連接數(shù)。2.3 業(yè)務(wù)主動上報HTTP SDK Kafka注意采樣與限流有些數(shù)據(jù)源既不是日志也不是數(shù)據(jù)庫而是客戶端行為埋點前端點擊、APP啟動等。這時候通常是業(yè)務(wù)方直接調(diào)用HTTP接口上報你在接口里把數(shù)據(jù)寫入Kafka。這個環(huán)節(jié)最常見的坑是突發(fā)流量打垮寫入服務(wù)。我在某個項目中遇到過前端埋點日志突然暴增導(dǎo)致上報接口被瞬間打滿Kafka客戶端批量發(fā)送超時丟了一批數(shù)據(jù)。后來做了三層保護SDK端批量發(fā)送不要一條一條發(fā)HTTP請求在SDK內(nèi)攢批比如攢夠100條或500ms顯著降低請求頻率服務(wù)端限流單機QPS上限設(shè)置好超出部分直接丟棄并記錄日志注意埋點數(shù)據(jù)丟幾條通常不影響大屏指標(biāo)趨勢但要保證不拖垮服務(wù)Kafka端分區(qū)數(shù)規(guī)劃根據(jù)峰值吞吐預(yù)估分區(qū)數(shù)分區(qū)數(shù) 目標(biāo)吞吐量 / 單分區(qū)吞吐量。例如目標(biāo)10萬條/秒單分區(qū)吞吐約2萬條/秒分區(qū)數(shù)至少5個數(shù)據(jù)采集層的通用規(guī)范也很重要。所有上報數(shù)據(jù)統(tǒng)一JSON格式包含event_id全局唯一、event_time事件發(fā)生時間、source數(shù)據(jù)來源、biz_body業(yè)務(wù)字段。有了這個規(guī)范后續(xù)Flink側(cè)做解析、去重、Watermark定義都有據(jù)可依。3. Flink實時計算核心狀態(tài)、時間語義與Sink的坑到了計算層就是Flink的主戰(zhàn)場。這里我把最高頻的三個技術(shù)點拆開講這三個點也是面試和實戰(zhàn)中最容易翻車的狀態(tài)管理、時間語義、自定義Sink。3.1 狀態(tài)與Checkpoint為什么你的作業(yè)重啟丟數(shù)據(jù)Flink的狀態(tài)State是它區(qū)別于普通流處理引擎的核心能力。簡單理解狀態(tài)就是算到一半的中間結(jié)果。比如你要統(tǒng)計每分鐘每個商品的累計銷售額這個累計值就需要保存下來這就是State。我見過很多使用者在應(yīng)用里定義了一個MapState來保存用戶維度的累計數(shù)據(jù)然后把Checkpoint間隔設(shè)置成5分鐘。結(jié)果某個凌晨Flink作業(yè)因為OOM掛掉了恢復(fù)后發(fā)現(xiàn)損失了將近10分鐘的統(tǒng)計結(jié)果。復(fù)盤時發(fā)現(xiàn)Checkpoint間隔太大狀態(tài)恢復(fù)點太靠前中間的數(shù)據(jù)全丟了。這里必須記住一個基本參數(shù)組合state.backend: rocksdb state.checkpoints.dir: hdfs:///flink/checkpoints execution.checkpointing.interval: 60s execution.checkpointing.mode: EXACTLY_ONCE execution.checkpointing.timeout: 5min我的經(jīng)驗是線上作業(yè)至少每分鐘做一次Checkpoint太頻繁會影響性能但5分鐘就太長了。另外一定要用RocksDB作為狀態(tài)后端——數(shù)據(jù)量一大純內(nèi)存Heap狀態(tài)分分鐘把JVM堆撐爆。RocksDB是把狀態(tài)寫到本地磁盤內(nèi)存只是緩存可靠性和容量都更好。3.2 事件時間與Watermark亂序數(shù)據(jù)怎么算Flink的窗口計算有個經(jīng)典三選一ProcessingTime、EventTime、IngestionTime。做實時大屏我強烈建議用EventTime也就是按業(yè)務(wù)事件發(fā)生的時間來劃分窗口。但EventTime帶來的問題是數(shù)據(jù)可能亂序到達。用戶點擊發(fā)生在10:00:00的日志可能到10:00:30才到Flink。如果你正好在做每分鐘點擊量的滾動窗口這條數(shù)據(jù)就會被算到10:01的窗口里指標(biāo)就錯了。解決方案是Watermark水位線它表示事件時間小于等于這個值的數(shù)據(jù)都已經(jīng)到達了。DataStreamOrderEvent withWatermark orders .assignTimestampsAndWatermarks( WatermarkStrategy.OrderEventforBoundedOutOfOrderness( Duration.ofSeconds(30) ) .withTimestampAssigner((event, timestamp) - event.getEventTime()) );forBoundedOutOfOrderness(Duration.ofSeconds(30))的意思是容忍最多30秒的亂序。代價是窗口結(jié)果會延遲30秒才輸出。這里就是業(yè)務(wù)延遲和數(shù)據(jù)準(zhǔn)確率的權(quán)衡。如果大屏指標(biāo)允許延遲30秒這個配置就合理如果要求秒級延遲那就要接受部分亂序數(shù)據(jù)會算錯窗口。窗口計算上我做實時指標(biāo)統(tǒng)計會用TumblingEventTimeWindows滾動窗口AllowedLateness的組合stream.keyBy(OrderEvent::getProductId) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .allowedLateness(Time.seconds(30)) .aggregate(new CountAggregate()) .process(new WindowResultFunction());allowedLateness的意思是窗口正常計算后還會等30秒的遲到數(shù)據(jù)遲到數(shù)據(jù)到達時單獨觸發(fā)一次計算輸出更新。這樣既保證了主鏈路結(jié)果快速產(chǎn)出又能修正部分亂序數(shù)據(jù)帶來的誤差。3.3 自定義DataSource與DataSink從入門到放棄再入門熱搜詞里flink 自定義 data source和flink 自定義 data sink出現(xiàn)頻率極高我猜是因為官方文檔的示例太簡單一上生產(chǎn)就漏出各種問題。這里我把兩個痛點講透。自定義DataSource通常是為了從非標(biāo)準(zhǔn)源讀數(shù)據(jù)。核心是繼承RichSourceFunction或?qū)崿F(xiàn)SourceFunctionpublic class MetricSource extends RichSourceFunctionMetricEvent { private volatile boolean running true; private transient KafkaProducer producer; Override public void open(Configuration parameters) { producer new KafkaProducer(...); } Override public void run(SourceContextMetricEvent ctx) throws Exception { while (running) { // 模擬讀取外部數(shù)據(jù)源 MetricEvent event readFromExternalSystem(); synchronized (ctx.getCheckpointLock()) { ctx.collect(event); } } } Override public void cancel() { running false; } }注意兩點一是collect操作必須在ctx.getCheckpointLock()鎖內(nèi)執(zhí)行否則Checkpoint時的狀態(tài)一致性會出問題數(shù)據(jù)可能重復(fù)或丟失。二是cancel()方法里要釋放外部連接資源否則作業(yè)取消時連接泄漏時間長了會把源系統(tǒng)連接池打滿。自定義DataSink的坑就更多了。我之前寫過自定義Sink寫入某個內(nèi)部監(jiān)控平臺代碼如下public class MonitorSink extends RichSinkFunctionMetricEvent { private MonitorClient client; Override public void open(Configuration parameters) { client MonitorClient.connect(monitor-server:8080); } Override public void invoke(MetricEvent value, Context context) throws Exception { boolean success client.send(value); if (!success) { throw new RuntimeException(send metric failed: value); } } Override public void close() { client.close(); } }這段代碼看起來沒問題生產(chǎn)上卻出過事故監(jiān)控平臺的單機處理能力有限Flink端并發(fā)寫入量一大client.send就頻繁超時我讓invoke直接拋異常結(jié)果Flink作業(yè)一直在重啟上游Kafka消費被阻滯整個實時鏈路癱瘓。3.4 從事故學(xué)到的Sink設(shè)計原則那次事故之后我給自己定了幾條Sink設(shè)計的鐵律也分享給你寫外部系統(tǒng)必須做重試和熔斷不能一失敗就拋異常重啟作業(yè)。應(yīng)該捕獲異常做有限次數(shù)重試重試仍失敗就寫本地容災(zāi)文件或者發(fā)告警跳過保證主鏈路不中斷區(qū)分業(yè)務(wù)錯誤和系統(tǒng)錯誤數(shù)據(jù)格式錯誤比如字段缺失屬于業(yè)務(wù)錯誤直接throw沒問題因為重試一萬次也還是會失敗外部系統(tǒng)不可用屬于系統(tǒng)錯誤應(yīng)該讓作業(yè)保留現(xiàn)場繼續(xù)運行等待外部系統(tǒng)恢復(fù)批量寫入優(yōu)先于逐條寫入能批量就別單條單條寫的性能開銷太大了。Flink提供了JdbcBatchingOutputFormat支持?jǐn)€批提交但要注意攢批參數(shù)batchSize和batchInterval要配合好另外熱搜詞里flink的jdbc連接器異常是個高頻問題。我遇到過的大部分情況是連接池耗盡和連接空閑超時。Flink JDBC Sink的每個并發(fā)Task都會建自己的連接你在連接池里配置了最大連接數(shù)10結(jié)果Flink作業(yè)并行度是20直接就有10個Task拿不到連接報錯。解決辦法很簡單要么把連接池最大連接數(shù)設(shè)成大于等于Flink并行度要么給連接設(shè)置合理的maxRetryTimes和connectionTimeout。4. 高頻事故復(fù)盤Flink Sink到Hive表數(shù)據(jù)不落盤的根因這一節(jié)我要重點復(fù)盤一個幾乎每個做Flink接數(shù)倉的人都會踩的坑——Flink sink Hive表數(shù)據(jù)不入表。這個熱搜詞出現(xiàn)得如此頻繁說明大家都在這上面栽過跟頭。我把排查鏈路完整還原出來你以后遇到可以直接照著查。4.1 現(xiàn)象與第一反應(yīng)當(dāng)時的情況是Flink作業(yè)運行狀態(tài)正常沒有報錯但查詢Hive表時發(fā)現(xiàn)數(shù)據(jù)一直是空的或者只有很久以前的一部分?jǐn)?shù)據(jù)。我第一反應(yīng)是是不是SQL寫錯了結(jié)果檢查Flink SQL和Table Schema都對得上Kafka source也在正常消費。于是開始逐步排查。4.2 排查鏈路四個層面逐個擊破第一層看Flink作業(yè)日志別被正常騙了打開TaskManager日志結(jié)果發(fā)現(xiàn)了端倪日志里出現(xiàn)了大量Need to partition the files into Hives format和Abortable相關(guān)的詞。這個信息很關(guān)鍵——Flink寫Hive是按照分區(qū)來管理的如果你沒有開啟自動提交分區(qū)數(shù)據(jù)寫入的是臨時目錄永遠(yuǎn)不會變成Hive的正式分區(qū)。第二層確認(rèn)Hive表的分區(qū)提交機制Flink寫Hive表默認(rèn)配置涉及兩個核心參數(shù)。如果你的Hive表是分區(qū)表必須顯式開啟分區(qū)提交并且設(shè)置正確的提交觸發(fā)策略CREATE TABLE hive_orders ( order_id BIGINT, product_id BIGINT, amount DOUBLE ) PARTITIONED BY (dt STRING, hour STRING) WITH ( connector hive, sink.partition-commit.trigger partition-time, sink.partition-commit.delay 0s, sink.partition-commit.policy.kind metastore,success-file );sink.partition-commit.trigger如果沒配或者配成process-time意味著Flink按數(shù)據(jù)到達時間來決定提交分區(qū)不是按數(shù)據(jù)的事件時間。我當(dāng)時就是用了process-time結(jié)果業(yè)務(wù)上凌晨的數(shù)據(jù)被算到了早上的分區(qū)等了一早上沒看到該有的數(shù)據(jù)。第三層檢查寫入文件格式與可見性很多剛用Flink寫Hive的同學(xué)不知道Flink寫Hive默認(rèn)是寫ORC或Parquet格式文件到分區(qū)的臨時目錄然后通過Table Metastore注冊分區(qū)。但文件從寫入中到可見之間有一個提交環(huán)節(jié)。如果你看到HDFS上分區(qū)目錄下已經(jīng)有Parquet文件但查詢不到數(shù)據(jù)大概率就是分區(qū)提交沒有正確執(zhí)行。還有一種可能是你寫的是非分區(qū)表Flink寫非分區(qū)表會把數(shù)據(jù)直接寫到表的目錄下。但我見過一個案例表本身是分區(qū)表Flink SQL里卻只指定了分區(qū)字段的部分值導(dǎo)致Sink端認(rèn)為這是一個不可寫分區(qū)就一直默默丟數(shù)據(jù)。排查方法是用SHOW PARTITIONS hive_orders看分區(qū)元數(shù)據(jù)是否存在。第四層Hive Streaming協(xié)議與Metastore對接Flink寫Hive底層有兩種協(xié)議一種是通用的Hive Streaming API通過HiveTableSink另一種是直接寫文件然后調(diào)用Metastore注冊分區(qū)。前者需要開啟hive.streaming.enabled老版本。如果你用的是較老版本的Flink和Hive建議用hive-streaming-client包并顯式開啟dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-hive_2.12/artifactId version你的版本/version /dependency注意Flink和Hive的版本兼容矩陣很挑剔Flink 1.14之前和Hive 3.1.0有兼容問題Hive Streaming API在部分版本上是broken的。最穩(wěn)妥的做法是Flink 1.15配Hive 3.1.2同時把Metastore從嵌入式切換為獨立部署避免并發(fā)寫入時Metastore鎖沖突。4.3 問題根因與修復(fù)最終我們定位到了根因Flink作業(yè)里寫Hive表的并行度設(shè)置過高默認(rèn)等于Kafka分區(qū)數(shù)每個并發(fā)Task都在嘗試寫同一個分區(qū)的臨時文件而且Flink內(nèi)部會出現(xiàn)文件內(nèi)容不完整的競態(tài)。配合開啟分區(qū)提交后問題消失。修復(fù)后的經(jīng)驗總結(jié)成一條Flink寫Hive表并行度不建議大于1。因為Hive表Sink的文件寫入不是天然按key分區(qū)的多個并行度同時寫一個分區(qū)文件合并和提交的復(fù)雜度會指數(shù)上升。要么做rebalance并設(shè)置并行度1要么就用bucket功能把數(shù)據(jù)按字段散列到多個文件讓Flink自己去整理。還有一個很小的點容易被忽略檢查你的作業(yè)是在本地IDEA跑還是集群跑。本地跑Flink時HDFS路徑如果寫的是hdfs://...會直接連不上集群如果寫的是本地路徑file://...那數(shù)據(jù)其實是寫到你個人電腦磁盤上Hive當(dāng)然查不到。這種環(huán)境不一致問題我遇到過不止一次。5. 可視化大屏的實時感數(shù)據(jù)刷新頻率、聚合策略與接口設(shè)計實時計算做完數(shù)據(jù)源源不斷寫入結(jié)果表了最后一步是可視化大屏。很多團隊在這里其實沒有技術(shù)問題但做出來的大屏看起來不實時——有實時數(shù)據(jù)卻沒有實時感。這節(jié)講講大屏之后端數(shù)據(jù)接口設(shè)計。5.1 大屏的數(shù)據(jù)不要直連數(shù)據(jù)庫查詢我剛做第一個實時大屏項目的時候犯過低級錯誤大屏前端每5秒輪詢一次數(shù)據(jù)庫原始明細(xì)表SQL里現(xiàn)場做SUM和GROUP BY。這種做法的結(jié)果是數(shù)據(jù)庫CPU飆升、查詢越來越慢、大屏的實時變成了每10秒才刷新一次因為查詢耗時就占了5秒。正確的思路是打一層結(jié)果表。Flink實時計算出來的指標(biāo)本來就已經(jīng)按分鐘/小時粒度聚合好了直接寫入結(jié)果表比如dashboard_metrics大屏端每5秒查詢的就只有幾行聚合好的數(shù)據(jù)查詢耗時基本在毫秒級。這才是實時大屏該有的性能。5.2 大屏接口的三種刷新模式輪詢模式前端每N秒調(diào)一次后端接口適合指標(biāo)值更新不頻繁、需要簡單穩(wěn)定的場景。N一般設(shè)為5秒或10秒。WebSocket推送Flink側(cè)結(jié)果更新時后端主動向已連接的大屏客戶端推送數(shù)據(jù)適合大屏數(shù)量多、希望即時刷新、減少無效請求的場景。SSE流推送如果你只做單向數(shù)據(jù)推送SSE比WebSocket更簡單基于HTTP協(xié)議兼容性和調(diào)試成本都低很多。這三個模式可以混合使用關(guān)鍵指標(biāo)用WebSocket推送次要指標(biāo)用輪詢兜底。前端技術(shù)棧我用得比較多的是Vue ECharts大屏布局用Grid實現(xiàn)自適應(yīng)。如果你不想花太多時間調(diào)布局可以直接用現(xiàn)成的DataV或者大屏編輯器但要注意編輯器的數(shù)據(jù)接入?yún)f(xié)議是否支持實時推送。5.3 減少大屏刷新壓力聚合結(jié)果表 緩存策略大屏本身有幾十個圖表如果每個圖表都單獨去查一次結(jié)果表也是壓力。我的做法是按業(yè)務(wù)場景把大屏所需的指標(biāo)打包成一個JSON大接口一次查詢返回所有圖表的數(shù)據(jù)。比如實時成交大屏這個場景接口返回的數(shù)據(jù)結(jié)構(gòu)大致是{ timestamp: 1715673600000, gmv: 102400.5, orderCount: 1287, userCount: 846, trend: [...], rankList: [...], geoDistribution: [...] }大屏端拿到這個JSON各自渲染對應(yīng)的圖表組件。這樣一個接口的查詢時間通常能控制在50ms以內(nèi)刷新頻率甚至可以提到1秒。還有一點給結(jié)果數(shù)據(jù)加Redis緩存。Flink寫入結(jié)果表的同時把熱數(shù)據(jù)同步一份到Redis大屏接口優(yōu)先讀Redis而不是查Doris或ClickHouse。Redis查詢是純內(nèi)存操作性能遠(yuǎn)高于OLAP數(shù)據(jù)庫。但要注意最終一致性——如果Flink寫入結(jié)果表成功但寫Redis失敗緩存里就是舊數(shù)據(jù)。我的解決方式是結(jié)果表帶一個update_time大屏接口拿數(shù)據(jù)時會比對Redis緩存時間和本地時間超過5秒則強制回源查結(jié)果表。5.4 大屏可視化的實時感還有視覺層面說實話大屏的實時感一半靠數(shù)據(jù)一半靠視覺設(shè)計。有幾位項目里的前端同學(xué)總結(jié)過一些經(jīng)驗非常有效數(shù)字跳動效果關(guān)鍵指標(biāo)成交額、訂單數(shù)用滾動數(shù)字代替靜態(tài)數(shù)字視覺上強化正在變化的感受刷新閃光提示每次刷新成功后給指標(biāo)卡片加一個淡入的閃爍效果說明我更新了時序圖的時間軸ECharts的時間軸坐標(biāo)保持固定寬度數(shù)據(jù)向右側(cè)推進給人一種趨勢正在流動的感覺最后更新時間顯示大屏角落永遠(yuǎn)顯示數(shù)據(jù)截至 HH:mm:ss讓使用者知道數(shù)據(jù)有多新這些都是細(xì)節(jié)但對于不懂技術(shù)的領(lǐng)導(dǎo)來說看起來實時和數(shù)據(jù)實時同等重要。6. 全鏈路延遲測量與容災(zāi)沒有指標(biāo)就沒有發(fā)言權(quán)實時平臺上線只是開始真正難的是讓它穩(wěn)定運行、出了問題能快速定位。這一節(jié)講講我怎么給實時鏈路做體檢和急救。6.1 延遲指標(biāo)每個環(huán)節(jié)都要有鐘表我構(gòu)建的任何實時平臺都會在數(shù)據(jù)流里埋一個端到端延遲衡量機制。思路很簡單在數(shù)據(jù)入口打上時間戳在每個關(guān)鍵節(jié)點記錄觀察時間。我在采集端會在每個事件的頭部塞一個ingest_time然后Flink計算層、存儲層、接口層分別在日志里記下當(dāng)前時間。通過一條測試數(shù)據(jù)就能算出延遲環(huán)節(jié)計算方式常見瓶頸采集延遲Kafka收到時間 - 事件發(fā)生時間日志攢批時間過長、Filebeat端阻塞傳輸延遲Flink收到時間 - Kafka收到時間Kafka broker配置、網(wǎng)絡(luò)帶寬計算延遲Flink輸出時間 - Flink收到時間窗口尺寸、狀態(tài)大小、反壓存儲延遲數(shù)據(jù)庫落庫時間 - Flink輸出時間Sink并行度、批量提交間隔展示延遲大屏收到時間 - 數(shù)據(jù)庫返回時間前端輪詢周期、接口查詢耗時實操里我會寫一個LatencyMonitor的Flink作業(yè)專門消費Kafka的監(jiān)控topic解析每個事件的ingest_time并計算延遲分布P50/P95/P99再寫入監(jiān)控面板。延遲一旦超過閾值就觸發(fā)告警。6.2 每個環(huán)節(jié)的容災(zāi)機制實時鏈路比離線鏈路脆弱得多任何一個環(huán)節(jié)抖動都會波及到后面。我的容災(zāi)設(shè)計分三層數(shù)據(jù)源頭采集端必須保證數(shù)據(jù)不丟。Filebeat有本地backlog機制Flink CDC有Binlog位點記錄Kafka有多副本。這三層可以保證即使整個實時平臺崩潰數(shù)據(jù)還在源端或Kafka里躺著。Flink作業(yè)打開Checkpoint配合RestartStrategy自動恢復(fù)。我常用的策略是fixed-delay3次重試間隔10秒。如果3次都失敗就不盲目重啟了發(fā)告警讓人工介入避免無限重啟導(dǎo)致狀態(tài)反復(fù)加載、Kafka消費位點反復(fù)跳躍的惡性循環(huán)。存儲與展示結(jié)果表要設(shè)計冪等寫入Flink重啟后重放數(shù)據(jù)不會產(chǎn)生重復(fù)數(shù)據(jù)。大屏端接口要做降級——如果結(jié)果表查詢失敗至少返回緩存數(shù)據(jù)或者數(shù)據(jù)暫不可用的明確提示而不是白屏。6.3 數(shù)據(jù)積壓是最大的坑三招止損實時鏈路最怕的故障就是數(shù)據(jù)積壓——Kafka里堆積了大量未消費的數(shù)據(jù)Flink作業(yè)無論如何都追不上這時你看到的大屏是越來越舊的數(shù)據(jù)實時性徹底丟失。數(shù)據(jù)積壓的典型原因和處理方式Flink作業(yè)遇到瓶頸看Flink UI的Backpressure指標(biāo)如果Source端顯示High/Medium說明是下游處理不過來需要增加并行度或優(yōu)化算子邏輯如果Sink端顯示High說明寫外部系統(tǒng)慢了需要檢查外部系統(tǒng)的連接池、批量參數(shù)。上游突然峰值流量比如大促秒殺采集量瞬間漲10倍。這時候Flink集群如果沒有彈性擴縮容只能硬扛。我的建議是Kafka的topic保留時間設(shè)長一點7天等峰值過去后Flink作業(yè)自動追趕消費。Sink端故障比如ClickHouse或Doris暫時不可用Flink的Sink會積壓數(shù)據(jù)在算子內(nèi)部。如果積壓太嚴(yán)重我在Doris不可用期間會臨時把結(jié)果寫到Kafka的另一個備份topic等Doris恢復(fù)后重放。數(shù)據(jù)積壓其實是實時平臺的急性病處理原則是先止損、后排查——先通過擴容或調(diào)整并行度把消費速度提上來再來分析瓶頸根因。反之如果先停下來查根因積壓只會越來越多雪上加霜。6.4 關(guān)于實時平臺到底需要多實時的一些個人體會做完整條鏈路我的體會是實時平臺的技術(shù)難點從來不是某個單一組件而是整個鏈路的平衡工程。很多時候你不需要追求極致的秒級延遲只要端到端控制在10秒內(nèi)大屏的體驗已經(jīng)相當(dāng)好了。為了那個極致實時你付出的代價可能是系統(tǒng)復(fù)雜度翻倍、穩(wěn)定性踩坑無數(shù)。給新手的最實際建議是先用最簡單的方案跑通全鏈路——Kafka Flink 結(jié)果表 大屏接口把延遲指標(biāo)測出來再針對瓶頸做優(yōu)化。不要一上來就上CDC、上高級狀態(tài)、上復(fù)雜窗口先把骨架立起來。數(shù)據(jù)和可視化這條路上跑通的那一刻獲得的成就感比任何理論推演都來得實在。希望這篇實戰(zhàn)經(jīng)驗貼能幫你少走幾步彎路。 Flink實時數(shù)據(jù)分析平臺實戰(zhàn)：從數(shù)據(jù)采集到可視化大屏
1. 從需求到架構(gòu)先想清楚實時到底意味著什么接到一個實時數(shù)據(jù)分析平臺的需求時我心里第一反應(yīng)不是寫Flink代碼而是先問對方一句話你說的實時是秒級、分鐘級還是小時級這個問題問出來很多需求就瞬間清晰了。我見過太多團隊一上來就鋪Flink集群、搞大屏結(jié)果做了三個月發(fā)現(xiàn)核心指標(biāo)延遲五分鐘就能滿足白費了一堆功夫。Java工程師做實時數(shù)據(jù)平臺有個天然優(yōu)勢Flink本身就是Java/Scala生態(tài)你熟悉的Spring Boot、Maven、JVM調(diào)優(yōu)經(jīng)驗全部能復(fù)用。相比Python系或者純SQL系的數(shù)據(jù)棧Java團隊啃Flink的上手成本要低得多。這篇實戰(zhàn)文章我就圍繞數(shù)據(jù)采集→實時計算→可視化大屏這條完整鏈路把從零搭建一個實時數(shù)據(jù)分析平臺的架構(gòu)思路、核心代碼、踩坑記錄都講透。1.1 先拆需求你的大屏是不是假實時大多數(shù)實時數(shù)據(jù)分析平臺的真實需求拆開來看無非三類指標(biāo)監(jiān)控類比如訂單量、成交額、在線用戶數(shù)要求秒級或分鐘級刷新行為分析類用戶點擊流、頁面路徑要求準(zhǔn)實時但允許一定延遲預(yù)警通知類比如異常流量、交易失敗率飆升要求延遲越低越好這三類需求對技術(shù)選型的影響完全不同。我之前遇到一個做污水處理可視化大屏的項目客戶說實時結(jié)果詳細(xì)了解才知道污水?dāng)?shù)據(jù)本身是五分鐘采集一次那你就算用Flink做到毫秒級計算也沒有意義瓶頸在采集端。反過來如果是電商大促的實時成交大屏每秒鐘都有成千上萬條訂單事件那你就需要認(rèn)真設(shè)計從采集到展示的每一層。所以第一步永遠(yuǎn)是做延遲預(yù)算端到端延遲 采集延遲 傳輸延遲 計算延遲 存儲延遲 展示刷新延遲。把每一項都列出來標(biāo)出可接受范圍后續(xù)所有技術(shù)決策都有依據(jù)。1.2 端到端鏈路的分層設(shè)計我做的實時數(shù)據(jù)分析平臺標(biāo)準(zhǔn)鏈路分五層層級組件選型職責(zé)采集層Filebeat / Flink CDC / HTTP SDK將日志、數(shù)據(jù)庫變更、業(yè)務(wù)事件統(tǒng)一送入消息隊列傳輸層Kafka削峰填谷、緩沖削流解耦采集與計算計算層Flink實時ETL、窗口聚合、狀態(tài)計算、規(guī)則匹配存儲層Doris / ClickHouse / Redis結(jié)果表存儲、維度數(shù)據(jù)緩存、大屏查詢加速展示層Vue ECharts / DataV可視化大屏、指標(biāo)卡片、趨勢圖表這套鏈路跟傳統(tǒng)的離線數(shù)倉最大的區(qū)別在于數(shù)據(jù)不是按天批量加工而是以事件流的方式持續(xù)流動。Flink跑在Kafka和存儲之間相當(dāng)于一個永不停止的計算引擎——上游數(shù)據(jù)來了就算算完就寫寫完后端到端延遲通?？刂圃诿爰?。關(guān)于架構(gòu)理念現(xiàn)階段我做項目基本直接采用Kappa架構(gòu)思路不再搭建Lambda架構(gòu)。Lambda那套實時鏈路離線鏈路雙跑、最終結(jié)果合并的方案維護成本太高兩套代碼邏輯要一致本身就是災(zāi)難?，F(xiàn)在Flink的流批一體能力已經(jīng)相當(dāng)成熟一套代碼可以同時跑實時和離線Kappa架構(gòu)足夠覆蓋絕大多數(shù)場景。1.3 為什么選Flink而不是Spark Streaming每次做技術(shù)選型都要面對這個問題。我的答案很直接如果你的場景需要事件時間處理、精確一次語義、豐富的狀態(tài)管理Flink是當(dāng)前最優(yōu)解。事件時間處理數(shù)據(jù)在網(wǎng)絡(luò)上傳輸會有延遲和亂序Flink的Watermark機制可以基于事件真正發(fā)生的時間進行計算而不是基于數(shù)據(jù)到達時間。這在處理日志類數(shù)據(jù)時尤其重要——用戶點擊發(fā)生在10:00:00但因為網(wǎng)絡(luò)抖動這條日志10:00:10才到如果你用處理時間計算就把這10秒的誤差算進指標(biāo)里了。精確一次語義Exactly-OnceFlink通過Checkpoint 兩階段提交保證即使任務(wù)崩潰恢復(fù)數(shù)據(jù)也不會重復(fù)或丟失。做交易類指標(biāo)時這是剛需。狀態(tài)管理Flink可以把中間結(jié)果存在內(nèi)存或RocksDB中實現(xiàn)跨事件的聚合計算比如統(tǒng)計每個用戶的累計訪問次數(shù)這是純SQL流處理引擎很難做好的。當(dāng)然Spark Streaming在吞吐量上和微批處理也有自己的優(yōu)勢但說實話真心追求實時性的場景Flink的靈活性和生態(tài)完整度更適合。更何況現(xiàn)在Flink CDC已經(jīng)是數(shù)據(jù)庫實時采集的事實標(biāo)準(zhǔn)配合Java開發(fā)效率很高。2. 數(shù)據(jù)采集層的工程落地三種來源一套規(guī)范數(shù)據(jù)采集是整個實時鏈路的起點也是臟活累活最多的地方。很多同學(xué)把精力都花在Flink計算邏輯上結(jié)果數(shù)據(jù)源沒管好后面計算、展示全是垃圾進垃圾出。這里我按來源類型分開講。2.1 日志類采集Filebeat Kafka是黃金組合服務(wù)端日志是最常見的實時數(shù)據(jù)來源。我通常用Filebeat做日志采集器它比Flume輕量太多部署就是解壓一個二進制文件配置也簡單filebeat.inputs: - type: filestream enabled: true paths: - /data/logs/*.log fields: app_name: order-service log_type: business output.kafka: hosts: [kafka1:9092, kafka2:9092, kafka3:9092] topic: app-order-log partition.round_robin: reachable_only: true這個配置看起來簡單但有幾個細(xì)節(jié)務(wù)必注意不要用filestream直接用Kafka producer consumer方式Filebeat自帶背壓機制Kafka不可用時會暫停讀取本地文件不會丟數(shù)據(jù)。這是它作為采集端的核心理由。fields里打上應(yīng)用名和日志類型標(biāo)簽后面Flink消費時可以根據(jù)這些字段路由到不同處理邏輯。每個應(yīng)用單獨一個topic或者至少按業(yè)務(wù)線分topic。我曾經(jīng)見過所有應(yīng)用混在一個topic里的架構(gòu)Flink消費端要做大量過濾還會互相影響消費速度非常痛苦。2.2 數(shù)據(jù)庫變更采集Flink CDC到底怎么部署熱搜詞里flink cdc pipeline部署和flink cdc安裝部署出現(xiàn)頻率很高說明這個方向已經(jīng)成了實時數(shù)據(jù)平臺的主流需求。Flink CDC基于數(shù)據(jù)庫日志Binlog/Redo Log捕獲變更不打業(yè)務(wù)表對業(yè)務(wù)系統(tǒng)零侵入。部署上有兩種形態(tài)形態(tài)一Flink CDC作為Source接入Flink作業(yè)DataStreamSourceString stream env .addSource( MySqlSource.Stringbuilder() .hostname(localhost) .port(3306) .databaseList(shop) .tableList(shop.t_order) .username(cdc_user) .password(cdc_pwd) .deserializer(new JsonDebeziumDeserializationSchema()) .build() ) .setParallelism(1);這種形態(tài)適合在Flink作業(yè)里實時消費數(shù)據(jù)庫變更。注意setParallelism(1)很關(guān)鍵因為單個MySQL實例的Binlog讀取是單線程的并行度設(shè)置高了反而會出問題。形態(tài)二Flink CDC Pipeline獨立部署如果你的目標(biāo)是數(shù)據(jù)庫實時同步到另一個存儲可以用Flink CDC Pipeline也就是之前的CDAS它基于Yaml配置就能完成整庫同步不需要寫一行Java代碼source: type: mysql hostname: localhost port: 3306 username: cdc_user password: cdc_pwd tables: shop\.* sink: type: doris fenodes: doris:8030 username: admin password: admin123Pipeline形態(tài)適合快速落地但是如果你想在同步過程中做數(shù)據(jù)加工比如字段映射、類型轉(zhuǎn)換、過濾還是寫Java代碼更靈活。我的建議是同步裸數(shù)據(jù)用Pipeline需要加工用源碼。關(guān)于Flink CDC最大的坑是存量數(shù)據(jù)與增量數(shù)據(jù)的一致性問題。Flink CDC默認(rèn)會先做一次全量快照再切換到Binlog增量這個過程對數(shù)據(jù)庫有一定壓力。建議在業(yè)務(wù)低峰期做首次同步并且監(jiān)控好源庫的IOPS和連接數(shù)。2.3 業(yè)務(wù)主動上報HTTP SDK Kafka注意采樣與限流有些數(shù)據(jù)源既不是日志也不是數(shù)據(jù)庫而是客戶端行為埋點前端點擊、APP啟動等。這時候通常是業(yè)務(wù)方直接調(diào)用HTTP接口上報你在接口里把數(shù)據(jù)寫入Kafka。這個環(huán)節(jié)最常見的坑是突發(fā)流量打垮寫入服務(wù)。我在某個項目中遇到過前端埋點日志突然暴增導(dǎo)致上報接口被瞬間打滿Kafka客戶端批量發(fā)送超時丟了一批數(shù)據(jù)。后來做了三層保護SDK端批量發(fā)送不要一條一條發(fā)HTTP請求在SDK內(nèi)攢批比如攢夠100條或500ms顯著降低請求頻率服務(wù)端限流單機QPS上限設(shè)置好超出部分直接丟棄并記錄日志注意埋點數(shù)據(jù)丟幾條通常不影響大屏指標(biāo)趨勢但要保證不拖垮服務(wù)Kafka端分區(qū)數(shù)規(guī)劃根據(jù)峰值吞吐預(yù)估分區(qū)數(shù)分區(qū)數(shù) 目標(biāo)吞吐量 / 單分區(qū)吞吐量。例如目標(biāo)10萬條/秒單分區(qū)吞吐約2萬條/秒分區(qū)數(shù)至少5個數(shù)據(jù)采集層的通用規(guī)范也很重要。所有上報數(shù)據(jù)統(tǒng)一JSON格式包含event_id全局唯一、event_time事件發(fā)生時間、source數(shù)據(jù)來源、biz_body業(yè)務(wù)字段。有了這個規(guī)范后續(xù)Flink側(cè)做解析、去重、Watermark定義都有據(jù)可依。3. Flink實時計算核心狀態(tài)、時間語義與Sink的坑到了計算層就是Flink的主戰(zhàn)場。這里我把最高頻的三個技術(shù)點拆開講這三個點也是面試和實戰(zhàn)中最容易翻車的狀態(tài)管理、時間語義、自定義Sink。3.1 狀態(tài)與Checkpoint為什么你的作業(yè)重啟丟數(shù)據(jù)Flink的狀態(tài)State是它區(qū)別于普通流處理引擎的核心能力。簡單理解狀態(tài)就是算到一半的中間結(jié)果。比如你要統(tǒng)計每分鐘每個商品的累計銷售額這個累計值就需要保存下來這就是State。我見過很多使用者在應(yīng)用里定義了一個MapState來保存用戶維度的累計數(shù)據(jù)然后把Checkpoint間隔設(shè)置成5分鐘。結(jié)果某個凌晨Flink作業(yè)因為OOM掛掉了恢復(fù)后發(fā)現(xiàn)損失了將近10分鐘的統(tǒng)計結(jié)果。復(fù)盤時發(fā)現(xiàn)Checkpoint間隔太大狀態(tài)恢復(fù)點太靠前中間的數(shù)據(jù)全丟了。這里必須記住一個基本參數(shù)組合state.backend: rocksdb state.checkpoints.dir: hdfs:///flink/checkpoints execution.checkpointing.interval: 60s execution.checkpointing.mode: EXACTLY_ONCE execution.checkpointing.timeout: 5min我的經(jīng)驗是線上作業(yè)至少每分鐘做一次Checkpoint太頻繁會影響性能但5分鐘就太長了。另外一定要用RocksDB作為狀態(tài)后端——數(shù)據(jù)量一大純內(nèi)存Heap狀態(tài)分分鐘把JVM堆撐爆。RocksDB是把狀態(tài)寫到本地磁盤內(nèi)存只是緩存可靠性和容量都更好。3.2 事件時間與Watermark亂序數(shù)據(jù)怎么算Flink的窗口計算有個經(jīng)典三選一ProcessingTime、EventTime、IngestionTime。做實時大屏我強烈建議用EventTime也就是按業(yè)務(wù)事件發(fā)生的時間來劃分窗口。但EventTime帶來的問題是數(shù)據(jù)可能亂序到達。用戶點擊發(fā)生在10:00:00的日志可能到10:00:30才到Flink。如果你正好在做每分鐘點擊量的滾動窗口這條數(shù)據(jù)就會被算到10:01的窗口里指標(biāo)就錯了。解決方案是Watermark水位線它表示事件時間小于等于這個值的數(shù)據(jù)都已經(jīng)到達了。DataStreamOrderEvent withWatermark orders .assignTimestampsAndWatermarks( WatermarkStrategy.OrderEventforBoundedOutOfOrderness( Duration.ofSeconds(30) ) .withTimestampAssigner((event, timestamp) - event.getEventTime()) );forBoundedOutOfOrderness(Duration.ofSeconds(30))的意思是容忍最多30秒的亂序。代價是窗口結(jié)果會延遲30秒才輸出。這里就是業(yè)務(wù)延遲和數(shù)據(jù)準(zhǔn)確率的權(quán)衡。如果大屏指標(biāo)允許延遲30秒這個配置就合理如果要求秒級延遲那就要接受部分亂序數(shù)據(jù)會算錯窗口。窗口計算上我做實時指標(biāo)統(tǒng)計會用TumblingEventTimeWindows滾動窗口AllowedLateness的組合stream.keyBy(OrderEvent::getProductId) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .allowedLateness(Time.seconds(30)) .aggregate(new CountAggregate()) .process(new WindowResultFunction());allowedLateness的意思是窗口正常計算后還會等30秒的遲到數(shù)據(jù)遲到數(shù)據(jù)到達時單獨觸發(fā)一次計算輸出更新。這樣既保證了主鏈路結(jié)果快速產(chǎn)出又能修正部分亂序數(shù)據(jù)帶來的誤差。3.3 自定義DataSource與DataSink從入門到放棄再入門熱搜詞里flink 自定義 data source和flink 自定義 data sink出現(xiàn)頻率極高我猜是因為官方文檔的示例太簡單一上生產(chǎn)就漏出各種問題。這里我把兩個痛點講透。自定義DataSource通常是為了從非標(biāo)準(zhǔn)源讀數(shù)據(jù)。核心是繼承RichSourceFunction或?qū)崿F(xiàn)SourceFunctionpublic class MetricSource extends RichSourceFunctionMetricEvent { private volatile boolean running true; private transient KafkaProducer producer; Override public void open(Configuration parameters) { producer new KafkaProducer(...); } Override public void run(SourceContextMetricEvent ctx) throws Exception { while (running) { // 模擬讀取外部數(shù)據(jù)源 MetricEvent event readFromExternalSystem(); synchronized (ctx.getCheckpointLock()) { ctx.collect(event); } } } Override public void cancel() { running false; } }注意兩點一是collect操作必須在ctx.getCheckpointLock()鎖內(nèi)執(zhí)行否則Checkpoint時的狀態(tài)一致性會出問題數(shù)據(jù)可能重復(fù)或丟失。二是cancel()方法里要釋放外部連接資源否則作業(yè)取消時連接泄漏時間長了會把源系統(tǒng)連接池打滿。自定義DataSink的坑就更多了。我之前寫過自定義Sink寫入某個內(nèi)部監(jiān)控平臺代碼如下public class MonitorSink extends RichSinkFunctionMetricEvent { private MonitorClient client; Override public void open(Configuration parameters) { client MonitorClient.connect(monitor-server:8080); } Override public void invoke(MetricEvent value, Context context) throws Exception { boolean success client.send(value); if (!success) { throw new RuntimeException(send metric failed: value); } } Override public void close() { client.close(); } }這段代碼看起來沒問題生產(chǎn)上卻出過事故監(jiān)控平臺的單機處理能力有限Flink端并發(fā)寫入量一大client.send就頻繁超時我讓invoke直接拋異常結(jié)果Flink作業(yè)一直在重啟上游Kafka消費被阻滯整個實時鏈路癱瘓。3.4 從事故學(xué)到的Sink設(shè)計原則那次事故之后我給自己定了幾條Sink設(shè)計的鐵律也分享給你寫外部系統(tǒng)必須做重試和熔斷不能一失敗就拋異常重啟作業(yè)。應(yīng)該捕獲異常做有限次數(shù)重試重試仍失敗就寫本地容災(zāi)文件或者發(fā)告警跳過保證主鏈路不中斷區(qū)分業(yè)務(wù)錯誤和系統(tǒng)錯誤數(shù)據(jù)格式錯誤比如字段缺失屬于業(yè)務(wù)錯誤直接throw沒問題因為重試一萬次也還是會失敗外部系統(tǒng)不可用屬于系統(tǒng)錯誤應(yīng)該讓作業(yè)保留現(xiàn)場繼續(xù)運行等待外部系統(tǒng)恢復(fù)批量寫入優(yōu)先于逐條寫入能批量就別單條單條寫的性能開銷太大了。Flink提供了JdbcBatchingOutputFormat支持?jǐn)€批提交但要注意攢批參數(shù)batchSize和batchInterval要配合好另外熱搜詞里flink的jdbc連接器異常是個高頻問題。我遇到過的大部分情況是連接池耗盡和連接空閑超時。Flink JDBC Sink的每個并發(fā)Task都會建自己的連接你在連接池里配置了最大連接數(shù)10結(jié)果Flink作業(yè)并行度是20直接就有10個Task拿不到連接報錯。解決辦法很簡單要么把連接池最大連接數(shù)設(shè)成大于等于Flink并行度要么給連接設(shè)置合理的maxRetryTimes和connectionTimeout。4. 高頻事故復(fù)盤Flink Sink到Hive表數(shù)據(jù)不落盤的根因這一節(jié)我要重點復(fù)盤一個幾乎每個做Flink接數(shù)倉的人都會踩的坑——Flink sink Hive表數(shù)據(jù)不入表。這個熱搜詞出現(xiàn)得如此頻繁說明大家都在這上面栽過跟頭。我把排查鏈路完整還原出來你以后遇到可以直接照著查。4.1 現(xiàn)象與第一反應(yīng)當(dāng)時的情況是Flink作業(yè)運行狀態(tài)正常沒有報錯但查詢Hive表時發(fā)現(xiàn)數(shù)據(jù)一直是空的或者只有很久以前的一部分?jǐn)?shù)據(jù)。我第一反應(yīng)是是不是SQL寫錯了結(jié)果檢查Flink SQL和Table Schema都對得上Kafka source也在正常消費。于是開始逐步排查。4.2 排查鏈路四個層面逐個擊破第一層看Flink作業(yè)日志別被正常騙了打開TaskManager日志結(jié)果發(fā)現(xiàn)了端倪日志里出現(xiàn)了大量Need to partition the files into Hives format和Abortable相關(guān)的詞。這個信息很關(guān)鍵——Flink寫Hive是按照分區(qū)來管理的如果你沒有開啟自動提交分區(qū)數(shù)據(jù)寫入的是臨時目錄永遠(yuǎn)不會變成Hive的正式分區(qū)。第二層確認(rèn)Hive表的分區(qū)提交機制Flink寫Hive表默認(rèn)配置涉及兩個核心參數(shù)。如果你的Hive表是分區(qū)表必須顯式開啟分區(qū)提交并且設(shè)置正確的提交觸發(fā)策略CREATE TABLE hive_orders ( order_id BIGINT, product_id BIGINT, amount DOUBLE ) PARTITIONED BY (dt STRING, hour STRING) WITH ( connector hive, sink.partition-commit.trigger partition-time, sink.partition-commit.delay 0s, sink.partition-commit.policy.kind metastore,success-file );sink.partition-commit.trigger如果沒配或者配成process-time意味著Flink按數(shù)據(jù)到達時間來決定提交分區(qū)不是按數(shù)據(jù)的事件時間。我當(dāng)時就是用了process-time結(jié)果業(yè)務(wù)上凌晨的數(shù)據(jù)被算到了早上的分區(qū)等了一早上沒看到該有的數(shù)據(jù)。第三層檢查寫入文件格式與可見性很多剛用Flink寫Hive的同學(xué)不知道Flink寫Hive默認(rèn)是寫ORC或Parquet格式文件到分區(qū)的臨時目錄然后通過Table Metastore注冊分區(qū)。但文件從寫入中到可見之間有一個提交環(huán)節(jié)。如果你看到HDFS上分區(qū)目錄下已經(jīng)有Parquet文件但查詢不到數(shù)據(jù)大概率就是分區(qū)提交沒有正確執(zhí)行。還有一種可能是你寫的是非分區(qū)表Flink寫非分區(qū)表會把數(shù)據(jù)直接寫到表的目錄下。但我見過一個案例表本身是分區(qū)表Flink SQL里卻只指定了分區(qū)字段的部分值導(dǎo)致Sink端認(rèn)為這是一個不可寫分區(qū)就一直默默丟數(shù)據(jù)。排查方法是用SHOW PARTITIONS hive_orders看分區(qū)元數(shù)據(jù)是否存在。第四層Hive Streaming協(xié)議與Metastore對接Flink寫Hive底層有兩種協(xié)議一種是通用的Hive Streaming API通過HiveTableSink另一種是直接寫文件然后調(diào)用Metastore注冊分區(qū)。前者需要開啟hive.streaming.enabled老版本。如果你用的是較老版本的Flink和Hive建議用hive-streaming-client包并顯式開啟dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-hive_2.12/artifactId version你的版本/version /dependency注意Flink和Hive的版本兼容矩陣很挑剔Flink 1.14之前和Hive 3.1.0有兼容問題Hive Streaming API在部分版本上是broken的。最穩(wěn)妥的做法是Flink 1.15配Hive 3.1.2同時把Metastore從嵌入式切換為獨立部署避免并發(fā)寫入時Metastore鎖沖突。4.3 問題根因與修復(fù)最終我們定位到了根因Flink作業(yè)里寫Hive表的并行度設(shè)置過高默認(rèn)等于Kafka分區(qū)數(shù)每個并發(fā)Task都在嘗試寫同一個分區(qū)的臨時文件而且Flink內(nèi)部會出現(xiàn)文件內(nèi)容不完整的競態(tài)。配合開啟分區(qū)提交后問題消失。修復(fù)后的經(jīng)驗總結(jié)成一條Flink寫Hive表并行度不建議大于1。因為Hive表Sink的文件寫入不是天然按key分區(qū)的多個并行度同時寫一個分區(qū)文件合并和提交的復(fù)雜度會指數(shù)上升。要么做rebalance并設(shè)置并行度1要么就用bucket功能把數(shù)據(jù)按字段散列到多個文件讓Flink自己去整理。還有一個很小的點容易被忽略檢查你的作業(yè)是在本地IDEA跑還是集群跑。本地跑Flink時HDFS路徑如果寫的是hdfs://...會直接連不上集群如果寫的是本地路徑file://...那數(shù)據(jù)其實是寫到你個人電腦磁盤上Hive當(dāng)然查不到。這種環(huán)境不一致問題我遇到過不止一次。5. 可視化大屏的實時感數(shù)據(jù)刷新頻率、聚合策略與接口設(shè)計實時計算做完數(shù)據(jù)源源不斷寫入結(jié)果表了最后一步是可視化大屏。很多團隊在這里其實沒有技術(shù)問題但做出來的大屏看起來不實時——有實時數(shù)據(jù)卻沒有實時感。這節(jié)講講大屏之后端數(shù)據(jù)接口設(shè)計。5.1 大屏的數(shù)據(jù)不要直連數(shù)據(jù)庫查詢我剛做第一個實時大屏項目的時候犯過低級錯誤大屏前端每5秒輪詢一次數(shù)據(jù)庫原始明細(xì)表SQL里現(xiàn)場做SUM和GROUP BY。這種做法的結(jié)果是數(shù)據(jù)庫CPU飆升、查詢越來越慢、大屏的實時變成了每10秒才刷新一次因為查詢耗時就占了5秒。正確的思路是打一層結(jié)果表。Flink實時計算出來的指標(biāo)本來就已經(jīng)按分鐘/小時粒度聚合好了直接寫入結(jié)果表比如dashboard_metrics大屏端每5秒查詢的就只有幾行聚合好的數(shù)據(jù)查詢耗時基本在毫秒級。這才是實時大屏該有的性能。5.2 大屏接口的三種刷新模式輪詢模式前端每N秒調(diào)一次后端接口適合指標(biāo)值更新不頻繁、需要簡單穩(wěn)定的場景。N一般設(shè)為5秒或10秒。WebSocket推送Flink側(cè)結(jié)果更新時后端主動向已連接的大屏客戶端推送數(shù)據(jù)適合大屏數(shù)量多、希望即時刷新、減少無效請求的場景。SSE流推送如果你只做單向數(shù)據(jù)推送SSE比WebSocket更簡單基于HTTP協(xié)議兼容性和調(diào)試成本都低很多。這三個模式可以混合使用關(guān)鍵指標(biāo)用WebSocket推送次要指標(biāo)用輪詢兜底。前端技術(shù)棧我用得比較多的是Vue ECharts大屏布局用Grid實現(xiàn)自適應(yīng)。如果你不想花太多時間調(diào)布局可以直接用現(xiàn)成的DataV或者大屏編輯器但要注意編輯器的數(shù)據(jù)接入?yún)f(xié)議是否支持實時推送。5.3 減少大屏刷新壓力聚合結(jié)果表 緩存策略大屏本身有幾十個圖表如果每個圖表都單獨去查一次結(jié)果表也是壓力。我的做法是按業(yè)務(wù)場景把大屏所需的指標(biāo)打包成一個JSON大接口一次查詢返回所有圖表的數(shù)據(jù)。比如實時成交大屏這個場景接口返回的數(shù)據(jù)結(jié)構(gòu)大致是{ timestamp: 1715673600000, gmv: 102400.5, orderCount: 1287, userCount: 846, trend: [...], rankList: [...], geoDistribution: [...] }大屏端拿到這個JSON各自渲染對應(yīng)的圖表組件。這樣一個接口的查詢時間通常能控制在50ms以內(nèi)刷新頻率甚至可以提到1秒。還有一點給結(jié)果數(shù)據(jù)加Redis緩存。Flink寫入結(jié)果表的同時把熱數(shù)據(jù)同步一份到Redis大屏接口優(yōu)先讀Redis而不是查Doris或ClickHouse。Redis查詢是純內(nèi)存操作性能遠(yuǎn)高于OLAP數(shù)據(jù)庫。但要注意最終一致性——如果Flink寫入結(jié)果表成功但寫Redis失敗緩存里就是舊數(shù)據(jù)。我的解決方式是結(jié)果表帶一個update_time大屏接口拿數(shù)據(jù)時會比對Redis緩存時間和本地時間超過5秒則強制回源查結(jié)果表。5.4 大屏可視化的實時感還有視覺層面說實話大屏的實時感一半靠數(shù)據(jù)一半靠視覺設(shè)計。有幾位項目里的前端同學(xué)總結(jié)過一些經(jīng)驗非常有效數(shù)字跳動效果關(guān)鍵指標(biāo)成交額、訂單數(shù)用滾動數(shù)字代替靜態(tài)數(shù)字視覺上強化正在變化的感受刷新閃光提示每次刷新成功后給指標(biāo)卡片加一個淡入的閃爍效果說明我更新了時序圖的時間軸ECharts的時間軸坐標(biāo)保持固定寬度數(shù)據(jù)向右側(cè)推進給人一種趨勢正在流動的感覺最后更新時間顯示大屏角落永遠(yuǎn)顯示數(shù)據(jù)截至 HH:mm:ss讓使用者知道數(shù)據(jù)有多新這些都是細(xì)節(jié)但對于不懂技術(shù)的領(lǐng)導(dǎo)來說看起來實時和數(shù)據(jù)實時同等重要。6. 全鏈路延遲測量與容災(zāi)沒有指標(biāo)就沒有發(fā)言權(quán)實時平臺上線只是開始真正難的是讓它穩(wěn)定運行、出了問題能快速定位。這一節(jié)講講我怎么給實時鏈路做體檢和急救。6.1 延遲指標(biāo)每個環(huán)節(jié)都要有鐘表我構(gòu)建的任何實時平臺都會在數(shù)據(jù)流里埋一個端到端延遲衡量機制。思路很簡單在數(shù)據(jù)入口打上時間戳在每個關(guān)鍵節(jié)點記錄觀察時間。我在采集端會在每個事件的頭部塞一個ingest_time然后Flink計算層、存儲層、接口層分別在日志里記下當(dāng)前時間。通過一條測試數(shù)據(jù)就能算出延遲環(huán)節(jié)計算方式常見瓶頸采集延遲Kafka收到時間 - 事件發(fā)生時間日志攢批時間過長、Filebeat端阻塞傳輸延遲Flink收到時間 - Kafka收到時間Kafka broker配置、網(wǎng)絡(luò)帶寬計算延遲Flink輸出時間 - Flink收到時間窗口尺寸、狀態(tài)大小、反壓存儲延遲數(shù)據(jù)庫落庫時間 - Flink輸出時間Sink并行度、批量提交間隔展示延遲大屏收到時間 - 數(shù)據(jù)庫返回時間前端輪詢周期、接口查詢耗時實操里我會寫一個LatencyMonitor的Flink作業(yè)專門消費Kafka的監(jiān)控topic解析每個事件的ingest_time并計算延遲分布P50/P95/P99再寫入監(jiān)控面板。延遲一旦超過閾值就觸發(fā)告警。6.2 每個環(huán)節(jié)的容災(zāi)機制實時鏈路比離線鏈路脆弱得多任何一個環(huán)節(jié)抖動都會波及到后面。我的容災(zāi)設(shè)計分三層數(shù)據(jù)源頭采集端必須保證數(shù)據(jù)不丟。Filebeat有本地backlog機制Flink CDC有Binlog位點記錄Kafka有多副本。這三層可以保證即使整個實時平臺崩潰數(shù)據(jù)還在源端或Kafka里躺著。Flink作業(yè)打開Checkpoint配合RestartStrategy自動恢復(fù)。我常用的策略是fixed-delay3次重試間隔10秒。如果3次都失敗就不盲目重啟了發(fā)告警讓人工介入避免無限重啟導(dǎo)致狀態(tài)反復(fù)加載、Kafka消費位點反復(fù)跳躍的惡性循環(huán)。存儲與展示結(jié)果表要設(shè)計冪等寫入Flink重啟后重放數(shù)據(jù)不會產(chǎn)生重復(fù)數(shù)據(jù)。大屏端接口要做降級——如果結(jié)果表查詢失敗至少返回緩存數(shù)據(jù)或者數(shù)據(jù)暫不可用的明確提示而不是白屏。6.3 數(shù)據(jù)積壓是最大的坑三招止損實時鏈路最怕的故障就是數(shù)據(jù)積壓——Kafka里堆積了大量未消費的數(shù)據(jù)Flink作業(yè)無論如何都追不上這時你看到的大屏是越來越舊的數(shù)據(jù)實時性徹底丟失。數(shù)據(jù)積壓的典型原因和處理方式Flink作業(yè)遇到瓶頸看Flink UI的Backpressure指標(biāo)如果Source端顯示High/Medium說明是下游處理不過來需要增加并行度或優(yōu)化算子邏輯如果Sink端顯示High說明寫外部系統(tǒng)慢了需要檢查外部系統(tǒng)的連接池、批量參數(shù)。上游突然峰值流量比如大促秒殺采集量瞬間漲10倍。這時候Flink集群如果沒有彈性擴縮容只能硬扛。我的建議是Kafka的topic保留時間設(shè)長一點7天等峰值過去后Flink作業(yè)自動追趕消費。Sink端故障比如ClickHouse或Doris暫時不可用Flink的Sink會積壓數(shù)據(jù)在算子內(nèi)部。如果積壓太嚴(yán)重我在Doris不可用期間會臨時把結(jié)果寫到Kafka的另一個備份topic等Doris恢復(fù)后重放。數(shù)據(jù)積壓其實是實時平臺的急性病處理原則是先止損、后排查——先通過擴容或調(diào)整并行度把消費速度提上來再來分析瓶頸根因。反之如果先停下來查根因積壓只會越來越多雪上加霜。6.4 關(guān)于實時平臺到底需要多實時的一些個人體會做完整條鏈路我的體會是實時平臺的技術(shù)難點從來不是某個單一組件而是整個鏈路的平衡工程。很多時候你不需要追求極致的秒級延遲只要端到端控制在10秒內(nèi)大屏的體驗已經(jīng)相當(dāng)好了。為了那個極致實時你付出的代價可能是系統(tǒng)復(fù)雜度翻倍、穩(wěn)定性踩坑無數(shù)。給新手的最實際建議是先用最簡單的方案跑通全鏈路——Kafka Flink 結(jié)果表 大屏接口把延遲指標(biāo)測出來再針對瓶頸做優(yōu)化。不要一上來就上CDC、上高級狀態(tài)、上復(fù)雜窗口先把骨架立起來。數(shù)據(jù)和可視化這條路上跑通的那一刻獲得的成就感比任何理論推演都來得實在。希望這篇實戰(zhàn)經(jīng)驗貼能幫你少走幾步彎路。