1. 從“流”與“批”的割裂說起為什么需要Flink如果你在過去幾年里接觸過大數(shù)據(jù)處理大概率聽說過Hadoop MapReduce和Apache Spark。MapReduce是批處理的鼻祖它將海量數(shù)據(jù)切分成塊分批處理穩(wěn)定但延遲高。Spark通過內(nèi)存計(jì)算和DAG執(zhí)行引擎極大地提升了批處理的性能并引入了微批Micro-batch的概念來處理流數(shù)據(jù)試圖用一個(gè)引擎統(tǒng)一批和流。然而微批的本質(zhì)依然是“批”。它把連續(xù)的數(shù)據(jù)流按照固定的時(shí)間窗口比如1秒切成一個(gè)個(gè)小批次然后對這些批次進(jìn)行批處理。這帶來了一個(gè)根本性問題延遲和準(zhǔn)確性的權(quán)衡。你想降低延遲就得把批次切得更小比如100毫秒但這會(huì)引入巨大的調(diào)度開銷系統(tǒng)吞吐量會(huì)急劇下降。更重要的是事件真正發(fā)生的時(shí)間Event Time和處理時(shí)間Processing Time之間存在漂移微批模型很難精確處理這種亂序事件導(dǎo)致計(jì)算結(jié)果不準(zhǔn)確。比如統(tǒng)計(jì)每分鐘的網(wǎng)站點(diǎn)擊量一個(gè)在59秒發(fā)生的點(diǎn)擊可能因?yàn)榫W(wǎng)絡(luò)延遲在下一分鐘的微批次里才被處理結(jié)果就被錯(cuò)誤地計(jì)入了下一分鐘。這種割裂催生了對真正的流處理的需求。我們需要一個(gè)系統(tǒng)它視數(shù)據(jù)為無界的流Unbounded Stream事件到來即處理并具備強(qiáng)大的狀態(tài)管理和事件時(shí)間處理能力能保證計(jì)算結(jié)果的準(zhǔn)確性和極低的延遲。這就是Apache Flink誕生的核心背景。它從一開始就被設(shè)計(jì)為一個(gè)有狀態(tài)的流計(jì)算引擎其“批處理”被視作“有界流”的一種特例。這種“流批一體”的架構(gòu)理念讓它在大數(shù)據(jù)實(shí)時(shí)處理領(lǐng)域脫穎而出。我第一次在生產(chǎn)環(huán)境接觸Flink是為了替換一個(gè)基于Spark Streaming的實(shí)時(shí)風(fēng)控系統(tǒng)。那個(gè)系統(tǒng)為了追求更低的延遲將微批間隔設(shè)到了500毫秒結(jié)果在業(yè)務(wù)高峰時(shí)段背壓Backpressure嚴(yán)重吞吐量完全跟不上還時(shí)常因?yàn)閬y序數(shù)據(jù)導(dǎo)致風(fēng)險(xiǎn)規(guī)則誤判。遷移到Flink后我們實(shí)現(xiàn)了真正的逐事件處理端到端延遲穩(wěn)定在100毫秒以內(nèi)并且利用其精確的事件時(shí)間窗口和Watermark機(jī)制徹底解決了亂序數(shù)據(jù)的計(jì)算準(zhǔn)確性問題。這讓我深刻體會(huì)到從“微批模擬流”到“原生流處理”并非簡單的性能提升而是一次架構(gòu)范式的根本轉(zhuǎn)變。2. Flink架構(gòu)核心當(dāng)一切皆流時(shí)引擎如何運(yùn)轉(zhuǎn)理解了“流優(yōu)先”的理念我們再來拆解Flink是如何實(shí)現(xiàn)它的。其架構(gòu)可以分三層來理解編程模型、運(yùn)行時(shí)引擎和部署模式。2.1 編程模型DataStream API與Table API/SQLFlink為開發(fā)者提供了不同抽象層次的編程接口。最底層、最靈活的是DataStream APIJava/Scala。它讓你能完全掌控?cái)?shù)據(jù)處理邏輯的每一個(gè)細(xì)節(jié)。你定義Source讀取數(shù)據(jù)經(jīng)過一系列Transformation如map,filter,keyBy,window最終由Sink寫出。這對于實(shí)現(xiàn)復(fù)雜的、定制化的流處理邏輯至關(guān)重要。例如實(shí)現(xiàn)一個(gè)自定義的窗口觸發(fā)器或者在狀態(tài)中維護(hù)一個(gè)復(fù)雜的機(jī)器學(xué)習(xí)模型。// 一個(gè)簡單的DataStream API示例統(tǒng)計(jì)每5秒內(nèi)每個(gè)用戶的點(diǎn)擊次數(shù) DataStreamClickEvent clicks env.addSource(new KafkaSource(...)); DataStreamTuple2String, Long result clicks .keyBy(event - event.userId) // 按用戶ID分組 .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 5秒滾動(dòng)事件時(shí)間窗口 .process(new ProcessWindowFunctionClickEvent, Tuple2String, Long, String, TimeWindow() { Override public void process(String key, Context context, IterableClickEvent elements, CollectorTuple2String, Long out) { long count 0; for (ClickEvent ignored : elements) { count; } out.collect(new Tuple2(key, count)); } });更高層的是Table API 和 SQL。這是Flink“流批一體”理念的直觀體現(xiàn)。你可以用標(biāo)準(zhǔn)的SQL或類SQL的Table API來編寫查詢Flink會(huì)自動(dòng)將其優(yōu)化并翻譯成底層的DataStream或DataSet批程序。這對于業(yè)務(wù)分析師和習(xí)慣聲明式編程的開發(fā)者非常友好能極大提升開發(fā)效率。CREATE TABLE語句可以定義一張表其數(shù)據(jù)源可能是一個(gè)Kafka流也可能是一個(gè)HDFS上的靜態(tài)文件但查詢語法是完全一致的。-- 使用Flink SQL實(shí)現(xiàn)同樣的功能 CREATE TABLE ClickEvents ( user_id STRING, event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL 5 SECOND ) WITH ( connector kafka, ... ); SELECT user_id, COUNT(*), TUMBLE_START(event_time, INTERVAL 5 SECOND) as win_start FROM ClickEvents GROUP BY user_id, TUMBLE(event_time, INTERVAL 5 SECOND);為什么要有兩層API這其實(shí)是權(quán)衡。Table API/SQL開發(fā)快、易于維護(hù)適合標(biāo)準(zhǔn)化的ETL和查詢業(yè)務(wù)。DataStream API則像“匯編語言”當(dāng)你需要極致優(yōu)化、實(shí)現(xiàn)非標(biāo)準(zhǔn)邏輯如復(fù)雜事件處理CEP或訪問底層狀態(tài)時(shí)它是唯一選擇。在實(shí)際項(xiàng)目中我們常?；旌鲜褂糜肧QL完成主要的業(yè)務(wù)邏輯再用DataStream API寫UDF用戶自定義函數(shù)來處理特殊需求。2.2 運(yùn)行時(shí)引擎JobManager、TaskManager與任務(wù)調(diào)度你的Flink程序Job提交后會(huì)在一個(gè)運(yùn)行時(shí)集群中執(zhí)行。這個(gè)集群主要由兩種進(jìn)程組成JobManagerJM 相當(dāng)于集群的“大腦”。每個(gè)Job有一個(gè)主導(dǎo)的JobManager。它負(fù)責(zé)接收J(rèn)obGraph 將你編寫的程序無論是DataStream還是SQL生成的編譯成一個(gè)由算子Operator頂點(diǎn)和數(shù)據(jù)流邊構(gòu)成的邏輯圖稱為JobGraph。調(diào)度任務(wù)Task 將JobGraph中的算子鏈Operator Chain優(yōu)化合并后拆分成具體的任務(wù)Task分配給TaskManager的任務(wù)槽Task Slot執(zhí)行。一個(gè)Task Slot是TM中資源調(diào)度的最小單元可以運(yùn)行一個(gè)或多個(gè)算子的子任務(wù)Subtask。協(xié)調(diào)檢查點(diǎn)Checkpoint 發(fā)起和協(xié)調(diào)所有任務(wù)進(jìn)行分布式快照這是Flink容錯(cuò)的核心。故障恢復(fù) 當(dāng)TaskManager或任務(wù)失敗時(shí)從最近的檢查點(diǎn)恢復(fù)狀態(tài)重新調(diào)度任務(wù)。TaskManagerTM 相當(dāng)于集群的“肌肉”。每個(gè)TM是一個(gè)JVM進(jìn)程負(fù)責(zé)執(zhí)行JobManager分配的任務(wù)。它包含一個(gè)或多個(gè)Task Slot。Slot的數(shù)量定義了TM的并發(fā)能力。一個(gè)Slot可以運(yùn)行一個(gè)完整的任務(wù)流水線如一個(gè)Source - Map - Sink的鏈這意味著同一個(gè)Slot內(nèi)的算子交換數(shù)據(jù)無需序列化和網(wǎng)絡(luò)傳輸效率極高。任務(wù)鏈Operator Chaining是Flink一個(gè)重要的優(yōu)化策略。Flink默認(rèn)會(huì)將并行度相同、且滿足轉(zhuǎn)發(fā)策略的算子例如map-filter鏈接在一起放在同一個(gè)線程Task中執(zhí)行。這減少了線程間切換和序列化/反序列化的開銷。但有時(shí)為了資源隔離或提高并行度比如keyBy后的算子需要網(wǎng)絡(luò)shuffle會(huì)強(qiáng)制斷開鏈你可能需要手動(dòng)禁用鏈化。注意 很多初學(xué)者在本地測試時(shí)感覺很快一上生產(chǎn)就慢往往忽略了Slot的資源分配。一個(gè)常見誤區(qū)是認(rèn)為一個(gè)Slot一個(gè)線程所以Slot越多越好。實(shí)際上你需要根據(jù)算子的并行度和鏈化情況來規(guī)劃Slot數(shù)量。如果Slot設(shè)置過多而任務(wù)鏈很少會(huì)導(dǎo)致大量線程空轉(zhuǎn)增加上下文切換開銷。通常建議Slot數(shù)量與CPU核心數(shù)保持合理關(guān)系并通過調(diào)整算子并行度來充分利用Slot。2.3 部署模式Session、Per-Job與ApplicationFlink提供了多種部署模式適應(yīng)不同場景Session模式 先啟動(dòng)一個(gè)長期運(yùn)行的Flink集群Session集群然后將多個(gè)Job提交到這個(gè)集群。優(yōu)點(diǎn)是資源共享提交Job快。缺點(diǎn)是“資源隔離”差一個(gè)Job的異常如OOM可能導(dǎo)致整個(gè)集群不穩(wěn)定影響其他Job。同時(shí)所有Job共用集群的類加載器可能存在依賴沖突。這適合對啟動(dòng)延遲敏感、且Job規(guī)模較小、運(yùn)行時(shí)間短的開發(fā)測試場景。Per-Job模式 為每個(gè)Job單獨(dú)啟動(dòng)一個(gè)Flink集群Job完成后集群釋放。優(yōu)點(diǎn)是資源隔離性好Job間互不影響類加載器也是隔離的。缺點(diǎn)是每個(gè)Job啟動(dòng)都需要申請資源、啟動(dòng)集群開銷較大。這適合生產(chǎn)環(huán)境中對穩(wěn)定性要求高、長期運(yùn)行的重要Job。Application模式 這是Per-Job模式的演進(jìn)。主要區(qū)別在于main()方法的執(zhí)行地點(diǎn)從客戶端移到了JobManager上。在Per-Job模式下客戶端需要執(zhí)行main()方法來生成JobGraph這意味著客戶端必須有完整的應(yīng)用依賴和配置。而在Application模式下你將整個(gè)應(yīng)用jar包提交給集群由JobManager來執(zhí)行main()方法。這極大地簡化了客戶端的部署特別適合基于Kubernetes或YARN的環(huán)境也避免了因客戶端與集群環(huán)境不一致導(dǎo)致的問題。這也是目前生產(chǎn)環(huán)境推薦的主流模式。如何選擇簡單來說開發(fā)測試用Session傳統(tǒng)的、對客戶端環(huán)境可控的生產(chǎn)作業(yè)可以用Per-Job而基于云原生或希望簡化運(yùn)維的強(qiáng)烈推薦Application模式。我們團(tuán)隊(duì)在Kubernetes上就全面采用了Application模式將Flink Job打包成Docker鏡像通過Helm Chart部署實(shí)現(xiàn)了完全的聲明式管理和資源隔離。3. 四大基石支撐Flink可靠、準(zhǔn)確運(yùn)行的關(guān)鍵機(jī)制如果說架構(gòu)是骨骼那么“四大基石”——時(shí)間、狀態(tài)、窗口和檢查點(diǎn)——就是讓Flink強(qiáng)大而可靠的肌肉和神經(jīng)。3.1 Time與Watermark在亂序世界中建立秩序流處理中時(shí)間有三種事件時(shí)間Event Time 事件實(shí)際發(fā)生的時(shí)間通常由數(shù)據(jù)本身的時(shí)間戳字段決定。這是最符合業(yè)務(wù)邏輯的時(shí)間概念。處理時(shí)間Processing Time 數(shù)據(jù)被Flink算子處理的系統(tǒng)時(shí)間。最簡單但結(jié)果不確定受系統(tǒng)負(fù)載和網(wǎng)絡(luò)延遲影響。攝入時(shí)間Ingestion Time 數(shù)據(jù)進(jìn)入Flink Source算子的時(shí)間。是事件時(shí)間和處理時(shí)間的折中能提供一定的順序保證且開銷比事件時(shí)間小。要使用事件時(shí)間就必須解決亂序問題。數(shù)據(jù)在傳輸過程中可能延遲或亂序到達(dá)。Watermark正是Flink用于衡量事件時(shí)間進(jìn)展、容忍亂序的機(jī)制。Watermark本質(zhì)上是一個(gè)特殊的時(shí)間戳它被插入到數(shù)據(jù)流中聲明“所有事件時(shí)間小于等于這個(gè)時(shí)間戳的事件理論上都應(yīng)該已經(jīng)到達(dá)了”。當(dāng)一個(gè)算子收到時(shí)間T的Watermark時(shí)它就可以認(rèn)為不會(huì)再收到比T更早或等于的數(shù)據(jù)了。例如設(shè)置一個(gè)最大亂序時(shí)間為2秒的Watermark策略。當(dāng)一個(gè)事件時(shí)間09:00:03的數(shù)據(jù)到達(dá)時(shí)Flink可能會(huì)生成一個(gè)09:00:013-2的Watermark。這意味著算子可以安全地對09:00:01之前的事件時(shí)間窗口進(jìn)行計(jì)算和關(guān)閉了。// 分配時(shí)間戳和生成Watermark以周期性生成器為例 DataStreamEvent stream env.addSource(...); DataStreamEvent withTimestampsAndWatermarks stream .assignTimestampsAndWatermarks( WatermarkStrategy.EventforBoundedOutOfOrderness(Duration.ofSeconds(2)) .withTimestampAssigner((event, timestamp) - event.getCreationTime()) );這里有一個(gè)關(guān)鍵的心得forBoundedOutOfOrderness中的延遲時(shí)間設(shè)置是一個(gè)業(yè)務(wù)和技術(shù)上的權(quán)衡。設(shè)得太小可能導(dǎo)致遲到數(shù)據(jù)被丟棄計(jì)算結(jié)果不準(zhǔn)確設(shè)得太大會(huì)導(dǎo)致窗口結(jié)果輸出延遲變長占用更多狀態(tài)存儲(chǔ)。你需要根據(jù)業(yè)務(wù)數(shù)據(jù)的亂序程度來合理設(shè)定。我們通常會(huì)先用一個(gè)較大的值如1分鐘上線通過監(jiān)控遲到數(shù)據(jù)Flink的side output可以捕獲遲到數(shù)據(jù)的數(shù)量逐步調(diào)整到一個(gè)最優(yōu)值。3.2 State讓流計(jì)算記住“過去”無狀態(tài)的流計(jì)算如單純的過濾、映射很簡單但價(jià)值有限。真正的業(yè)務(wù)邏輯往往需要“記憶”比如累計(jì)銷售額、去重、模式匹配。Flink的狀態(tài)State就是算子的記憶。Flink的狀態(tài)分為兩種算子狀態(tài)Operator State 狀態(tài)與一個(gè)算子的并行實(shí)例綁定。例如Kafka Source需要記錄每個(gè)分區(qū)消費(fèi)到的偏移量這就是算子狀態(tài)。當(dāng)算子并行度改變時(shí)狀態(tài)需要被重新分配邏輯相對復(fù)雜。鍵控狀態(tài)Keyed State 這是最常用、功能最強(qiáng)大的狀態(tài)。它與數(shù)據(jù)流中定義的Key通過keyBy()產(chǎn)生綁定。每個(gè)Key對應(yīng)一個(gè)獨(dú)立的狀態(tài)值。因?yàn)镵eyBy保證了相同Key的數(shù)據(jù)總是路由到同一個(gè)算子子任務(wù)所以鍵控狀態(tài)的訪問和更新非常高效。Flink提供了豐富的鍵控狀態(tài)類型ValueStateT單個(gè)值、ListStateT列表、MapStateUK, UV映射、ReducingStateT聚合等。// 使用ValueState實(shí)現(xiàn)一個(gè)簡單的去重相同key在一分鐘內(nèi)只輸出第一條 public class DeduplicateFunction extends KeyedProcessFunctionString, Event, Event { private transient ValueStateLong lastSeenState; Override public void open(Configuration parameters) { ValueStateDescriptorLong descriptor new ValueStateDescriptor(lastSeen, Long.class); lastSeenState getRuntimeContext().getState(descriptor); } Override public void processElement(Event value, Context ctx, CollectorEvent out) throws Exception { Long lastSeen lastSeenState.value(); long currentTime ctx.timestamp(); // 事件時(shí)間 if (lastSeen null || (currentTime - lastSeen 60000)) { // 一分鐘內(nèi)未出現(xiàn) lastSeenState.update(currentTime); out.collect(value); } } }狀態(tài)后端State Backend決定了狀態(tài)存儲(chǔ)在哪里、如何訪問。主要有三種HashMapStateBackend 狀態(tài)存儲(chǔ)在JVM堆內(nèi)存中。速度快但狀態(tài)大小受限于TaskManager內(nèi)存且Checkpoint時(shí)狀態(tài)會(huì)序列化存儲(chǔ)到分布式文件系統(tǒng)如HDFS。適合狀態(tài)小、對性能要求極高的場景。EmbeddedRocksDBStateBackend 狀態(tài)存儲(chǔ)在本地磁盤的RocksDB數(shù)據(jù)庫中TM進(jìn)程內(nèi)。支持的狀態(tài)量遠(yuǎn)大于內(nèi)存僅受磁盤限制并且Checkpoint時(shí)是增量快照效率高。但讀寫速度比內(nèi)存慢。這是生產(chǎn)環(huán)境最常用的選擇因?yàn)樗诖鬆顟B(tài)和性能之間取得了很好的平衡。FsStateBackend已逐漸被前兩者替代 一個(gè)折中方案狀態(tài)快照存儲(chǔ)于文件系統(tǒng)。選擇狀態(tài)后端時(shí)核心考量是狀態(tài)大小和訪問延遲。我們有一個(gè)實(shí)時(shí)用戶畫像更新的Job狀態(tài)大小超過500GB使用RocksDB后端運(yùn)行非常穩(wěn)定。如果換成HashMapTM早就OOM了。3.3 Window在無界流上定義有界計(jì)算窗口是將無界流數(shù)據(jù)劃分為有限塊進(jìn)行處理的核心抽象。Flink的窗口機(jī)制非常靈活主要分為兩類時(shí)間窗口Time Window 按時(shí)間劃分。這是最常用的。滾動(dòng)窗口Tumbling 窗口大小固定不重疊。如每5分鐘統(tǒng)計(jì)一次?；瑒?dòng)窗口Sliding 窗口大小固定但可以滑動(dòng)有重疊。如每1分鐘統(tǒng)計(jì)一次過去5分鐘的數(shù)據(jù)。會(huì)話窗口Session 根據(jù)活動(dòng)的非活躍間隙Gap來劃分窗口。非常適合用戶行為分析。計(jì)數(shù)窗口Count Window 按元素個(gè)數(shù)劃分。如每1000個(gè)點(diǎn)擊統(tǒng)計(jì)一次。窗口的核心組件包括窗口分配器Window Assigner 決定一個(gè)數(shù)據(jù)元素該被分配到哪個(gè)/哪些窗口。觸發(fā)器Trigger 決定一個(gè)窗口何時(shí)被計(jì)算觸發(fā)和清除。除了默認(rèn)的時(shí)間/計(jì)數(shù)觸發(fā)你可以自定義比如“收到特定事件時(shí)觸發(fā)”。驅(qū)逐器Evictor 在觸發(fā)器觸發(fā)后、計(jì)算前/后可以選擇性地移除窗口中的某些元素。一個(gè)高級技巧是使用遲到數(shù)據(jù)處理。即使有Watermark仍可能有數(shù)據(jù)在窗口關(guān)閉后才到達(dá)遲到數(shù)據(jù)。Flink允許你通過.sideOutputLateData()將遲到數(shù)據(jù)輸出到側(cè)輸出流Side Output然后進(jìn)行額外處理比如更新之前的結(jié)果或者記錄到日志中用于監(jiān)控和調(diào)優(yōu)Watermark策略。3.4 Checkpoint與Savepoint容錯(cuò)與版本管理的利器這是Flink高可靠性的基石。檢查點(diǎn)Checkpoint是Flink自動(dòng)、定期觸發(fā)的分布式快照用于故障恢復(fù)。它捕獲所有算子的狀態(tài)State以及數(shù)據(jù)流中的位置如Kafka偏移量。其核心算法是Chandy-Lamport異步屏障快照算法。簡單來說JobManager會(huì)周期性地向所有Source算子注入一個(gè)特殊的“屏障Barrier”標(biāo)記這個(gè)標(biāo)記隨著數(shù)據(jù)流向下游傳播。當(dāng)算子收到所有輸入流的屏障時(shí)就會(huì)對自己的狀態(tài)做一次快照。所有算子的快照完成后就形成了一個(gè)全局一致的檢查點(diǎn)。Savepoint與Checkpoint在技術(shù)上類似但目的不同。Savepoint是用戶手動(dòng)觸發(fā)的、全局一致的狀態(tài)快照主要用于有狀態(tài)的應(yīng)用程序升級 更新Flink版本或作業(yè)邏輯代碼后可以從Savepoint恢復(fù)狀態(tài)實(shí)現(xiàn)“熱更新”。集群遷移或擴(kuò)縮容。暫停和重啟應(yīng)用。注意 Checkpoint是輕量級的、自動(dòng)的設(shè)計(jì)目標(biāo)是快速恢復(fù)其元數(shù)據(jù)可能被后續(xù)的Checkpoint覆蓋。Savepoint是重量級的、手動(dòng)管理的設(shè)計(jì)目標(biāo)是長期存儲(chǔ)和版本化管理必須顯式創(chuàng)建和刪除。生產(chǎn)環(huán)境中我們通常會(huì)配置每分鐘一次的Checkpoint并在每次發(fā)布新版本前通過命令行或REST API手動(dòng)創(chuàng)建一個(gè)Savepoint。4. 從開發(fā)到部署一個(gè)完整Flink應(yīng)用的生命周期了解了核心概念我們來看如何讓一個(gè)Flink應(yīng)用跑起來。這里以一個(gè)經(jīng)典的實(shí)時(shí)數(shù)據(jù)ETL和聚合場景為例從Kafka讀取用戶行為日志清洗過濾后按用戶維度統(tǒng)計(jì)每分鐘的活躍度并將結(jié)果寫入MySQL和Kafka以供下游使用。4.1 環(huán)境準(zhǔn)備與依賴管理首先你需要一個(gè)Flink環(huán)境。對于本地學(xué)習(xí)和測試最簡單的方式是下載Flink的二進(jìn)制發(fā)行版解壓后運(yùn)行./bin/start-cluster.shLinux/Mac或bin\start-cluster.batWindows一個(gè)單機(jī)Session集群就啟動(dòng)了。訪問http://localhost:8081可以看到Web UI。對于生產(chǎn)環(huán)境通常部署在YARN或Kubernetes上。以YARN為例你需要一個(gè)Hadoop集群并確保Flink的Hadoop集成jar包在FLINK_HOME/lib目錄下。然后可以通過./bin/flink run -m yarn-cluster ...提交作業(yè)。依賴管理是第一個(gè)坑。Flink應(yīng)用通常需要連接器如flink-connector-kafka、格式如flink-json等依賴。必須注意依賴沖突特別是與Flink自身庫的沖突。最佳實(shí)踐是使用Maven Shade Plugin或Gradle Shadow Plugin將你的應(yīng)用及其所有依賴排除Flink核心庫打包成一個(gè)“胖JarFat Jar/Uber Jar”。在打包時(shí)務(wù)必使用scopeprovided/scope標(biāo)記Flink核心依賴如flink-java,flink-streaming-java因?yàn)樗鼈円呀?jīng)在集群中提供了。!-- Maven pom.xml 示例片段 -- dependencies !-- Flink核心依賴scope為provided -- dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency !-- 應(yīng)用需要的連接器和格式依賴打包進(jìn)fat jar -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-json/artifactId version${flink.version}/version /dependency dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.33/version /dependency /dependencies build plugins plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.2.4/version executions execution phasepackage/phase goals goalshade/goal /goals configuration createDependencyReducedPomfalse/createDependencyReducedPom artifactSet excludes !-- 排除已在集群中的依賴 -- excludeorg.apache.flink:*/exclude excludecom.google.code.findbugs:jsr305/exclude /excludes /artifactSet filters filter !-- 解決META-INF/services文件沖突 -- artifact*:*/artifact excludes excludeMETA-INF/*.SF/exclude excludeMETA-INF/*.DSA/exclude excludeMETA-INF/*.RSA/exclude /excludes /filter /filters transformers transformer implementationorg.apache.maven.plugins.shade.resource.ServicesResourceTransformer/ /transformers /configuration /execution /executions /plugin /plugins /build4.2 核心邏輯開發(fā)Source、Transformation與Sink接下來是編碼。我們使用DataStream API和Table API混合的方式。步驟一定義數(shù)據(jù)源Source我們使用Flink Kafka Connector。注意要選擇正確的Kafka版本。// DataStream API方式 Properties kafkaProps new Properties(); kafkaProps.setProperty(bootstrap.servers, kafka-broker:9092); kafkaProps.setProperty(group.id, flink-user-behavior-group); FlinkKafkaConsumerString kafkaConsumer new FlinkKafkaConsumer( user_behavior_topic, new SimpleStringSchema(), kafkaProps ); // 設(shè)置從最新偏移量開始消費(fèi)生產(chǎn)環(huán)境通常設(shè)置為從group.id記錄的偏移量開始 kafkaConsumer.setStartFromLatest(); DataStreamString kafkaStream env.addSource(kafkaConsumer);步驟二數(shù)據(jù)轉(zhuǎn)換Transformation先解析JSON字符串然后進(jìn)行過濾和轉(zhuǎn)換。// 1. 解析JSON DataStreamUserBehaviorEvent parsedStream kafkaStream .map(new MapFunctionString, UserBehaviorEvent() { Override public UserBehaviorEvent map(String value) throws Exception { ObjectMapper mapper new ObjectMapper(); return mapper.readValue(value, UserBehaviorEvent.class); } }) .returns(TypeInformation.of(UserBehaviorEvent.class)); // 顯式指定類型信息 // 2. 過濾無效數(shù)據(jù) DataStreamUserBehaviorEvent filteredStream parsedStream.filter(event - event.isValid()); // 3. 轉(zhuǎn)換為Table進(jìn)行聚合使用Table API // 首先創(chuàng)建表環(huán)境 StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); // 將DataStream注冊為一張臨時(shí)視圖 tableEnv.createTemporaryView(UserBehavior, filteredStream, Schema.newBuilder() .column(userId, DataTypes.STRING()) .column(behavior, DataTypes.STRING()) .column(timestamp, DataTypes.BIGINT()) .columnByExpression(ts, TO_TIMESTAMP_LTZ(timestamp, 3)) // 轉(zhuǎn)換時(shí)間戳 .watermark(ts, ts - INTERVAL 5 SECOND) // 定義Watermark .build()); // 執(zhí)行SQL查詢統(tǒng)計(jì)每分鐘每個(gè)用戶的活躍事件數(shù) Table resultTable tableEnv.sqlQuery( SELECT userId, COUNT(*) as activity_count, TUMBLE_START(ts, INTERVAL 1 MINUTE) as window_start, TUMBLE_END(ts, INTERVAL 1 MINUTE) as window_end FROM UserBehavior WHERE behavior IN (click, view, purchase) GROUP BY userId, TUMBLE(ts, INTERVAL 1 MINUTE) ); // 將Table轉(zhuǎn)換回DataStream以便后續(xù)處理 DataStreamResult resultStream tableEnv.toDataStream(resultTable, Result.class);步驟三數(shù)據(jù)輸出Sink結(jié)果需要寫入MySQL和Kafka。Flink提供了JDBC Sink和Kafka Sink。// 1. 寫入MySQL (使用JDBC Sink) JdbcExecutionOptions execOptions JdbcExecutionOptions.builder() .withBatchSize(1000) // 每批最多1000條 .withBatchIntervalMs(200) // 每200毫秒或批滿時(shí)刷出 .withMaxRetries(3) .build(); JdbcConnectionOptions connOptions new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl(jdbc:mysql://mysql-host:3306/rt_db) .withDriverName(com.mysql.cj.jdbc.Driver) .withUsername(user) .withPassword(pass) .build(); resultStream.addSink(JdbcSink.sink( INSERT INTO user_minute_activity (user_id, activity_count, window_start, window_end) VALUES (?, ?, ?, ?) ON DUPLICATE KEY UPDATE activity_count ?, (ps, t) - { ps.setString(1, t.userId); ps.setLong(2, t.activityCount); ps.setTimestamp(3, Timestamp.from(t.windowStart.toInstant())); ps.setTimestamp(4, Timestamp.from(t.windowEnd.toInstant())); ps.setLong(5, t.activityCount); // 用于ON DUPLICATE KEY UPDATE }, execOptions, connOptions )).name(jdbc-sink-mysql); // 2. 同時(shí)寫入Kafka供下游消費(fèi)如實(shí)時(shí)大屏 resultStream.map(result - result.toString()) // 轉(zhuǎn)換為字符串 .addSink(new FlinkKafkaProducer( result_topic, new SimpleStringSchema(), kafkaProps )).name(kafka-sink-result);4.3 配置、打包與提交開發(fā)完成后需要在main方法中配置執(zhí)行環(huán)境并設(shè)置關(guān)鍵的運(yùn)行時(shí)參數(shù)。public class UserBehaviorAnalysisJob { public static void main(String[] args) throws Exception { // 1. 創(chuàng)建流執(zhí)行環(huán)境 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // 生產(chǎn)環(huán)境建議明確設(shè)置并行度而不是用默認(rèn)值 env.setParallelism(4); // 2. 啟用Checkpoint (生產(chǎn)環(huán)境必須) env.enableCheckpointing(60000); // 每60秒一次 // 使用文件系統(tǒng)狀態(tài)后端路徑為HDFS或S3等持久化存儲(chǔ) env.setStateBackend(new EmbeddedRocksDBStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://namenode:8020/flink/checkpoints); // 設(shè)置精確一次語義 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 最小間隔防止過頻 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); // 超時(shí)時(shí)間 env.getCheckpointConfig().setCheckpointTimeout(600000); // 最大并發(fā)檢查點(diǎn)數(shù)量 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // 容忍的連續(xù)失敗次數(shù) env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); // 3. 設(shè)置重啟策略 env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 3, // 嘗試重啟次數(shù) Time.of(10, TimeUnit.SECONDS) // 重啟間隔 )); // 4. 組裝任務(wù)拓?fù)?(調(diào)用上面定義的source, transformation, sink邏輯) // ... // 5. 執(zhí)行任務(wù) env.execute(Real-time User Behavior Analysis); } }使用Maven打包mvn clean package -DskipTests。會(huì)在target目錄下生成一個(gè)your-app-1.0-SNAPSHOT.jar的胖Jar。提交到Y(jié)ARNApplication模式./bin/flink run-application -t yarn-application \ -Djobmanager.memory.process.size2048m \ -Dtaskmanager.memory.process.size4096m \ -Dtaskmanager.numberOfTaskSlots2 \ -Dyarn.application.nameFlink-UserBehavior-Analysis \ -c com.yourcompany.UserBehaviorAnalysisJob \ /path/to/your-app-1.0-SNAPSHOT.jar提交后可以在YARN ResourceManager UI和Flink Web UI上監(jiān)控作業(yè)的運(yùn)行狀態(tài)、背壓、Checkpoint情況等。4.4 生產(chǎn)環(huán)境運(yùn)維要點(diǎn)作業(yè)上線只是開始運(yùn)維監(jiān)控同樣重要。監(jiān)控指標(biāo) Flink提供了豐富的Metric通過Web UI、REST API或?qū)覲rometheus等監(jiān)控系統(tǒng)收集。關(guān)鍵指標(biāo)包括numRecordsIn/Out吞吐量、currentSendTime延遲、checkpointDuration檢查點(diǎn)耗時(shí)、lastCheckpointSize狀態(tài)大小、isBackPressured背壓等。日志管理 確保TaskManager和JobManager的日志被收集到中心化系統(tǒng)如ELK中便于排查問題。反壓Backpressure診斷 在Web UI的作業(yè)圖上如果某個(gè)節(jié)點(diǎn)顯示為紅色或橙色表示該節(jié)點(diǎn)正在經(jīng)歷反壓。原因可能是下游算子處理慢、數(shù)據(jù)傾斜、外部Sink如MySQL寫入慢等。需要結(jié)合Metrics和日志定位瓶頸。狀態(tài)調(diào)優(yōu) 對于RocksDB狀態(tài)后端可以調(diào)整state.backend.rocksdb前綴的配置如writebuffer.size,block.cache-size等以優(yōu)化讀寫性能。對于超大狀態(tài)可以考慮啟用增量Checkpoint和本地恢復(fù)。優(yōu)雅停止與升級 使用Savepoint進(jìn)行有狀態(tài)升級。流程是1) 使用stop --savepointPath ...停止當(dāng)前作業(yè)并觸發(fā)Savepoint2) 更新代碼并打包新Jar3) 使用run -s ...從Savepoint恢復(fù)啟動(dòng)新作業(yè)。從我的經(jīng)驗(yàn)看Flink作業(yè)上線后最常遇到的問題就是數(shù)據(jù)傾斜和外部系統(tǒng)連接。數(shù)據(jù)傾斜會(huì)導(dǎo)致個(gè)別Task負(fù)載極高成為瓶頸。解決方法包括在keyBy前對key加鹽打散或使用rebalance()強(qiáng)制均勻分發(fā)。外部系統(tǒng)連接如JDBC Sink則要注意連接池管理和批量寫入避免對數(shù)據(jù)庫造成過大壓力同時(shí)要處理好冪等性如上例中的ON DUPLICATE KEY UPDATE。