Flink知识点（二）｜Flink中是怎么处理乱序数据的

在 Flink 里“乱序”本质是事件时间event time先后顺序和到达时间processing time不一致。Flink 处理乱序数据的核心机制主要围绕事件时间语义 Watermark 窗口触发/延迟 迟到数据处理 状态与一致性。一、时间语义Flink中有三种时间语义​Processing Time按机器当前时间处理最简单、吞吐高但对乱序不敏感乱序会直接导致窗口统计不准。​Event Time最常用​按事件自带时间戳处理Flink 用Watermark推进“事件时间进度”从而容忍一定乱序。Ingestion Time介于两者之间进入 Flink 时打时间戳现在用得较少。乱序处理几乎都建立在Event Time上。‍二、WatermarkWatermark是乱序容忍与“事件时间进度”的关键。可以理解为“我认为未来不会再来时间戳 ≤ watermark的事件了”近似判断允许误差。常见生成策略​有界乱序(Bounded Out-of-Orderness)最常用设定最大乱序程度maxOutOfOrderness例如 5swatermark ≈ 当前观测到的最大事件时间 - 5s→ 能容忍 5 秒内乱序超过就可能变成迟到数据。​单调递增Monotonous Timestamps适合源数据严格按时间递增基本无乱序。​自定义 WatermarkGenerator适合多分区、多来源、需要特殊规则比如按业务字段分组、按分区对齐等。关键点Watermark 不是“等数据到齐”而是“推进时间并触发计算”的机制。对于并行 source下游算子的 watermark 通常取各输入分区 watermark 的最小值因此某个分区卡住会拖慢整体事件时间推进。‍三、窗口Window与触发Trigger乱序数据最典型场景是做窗口聚合滚动、滑动、会话窗口。Flink 的窗口何时“关窗”默认基于事件时间时当watermarkwindow_end时触发窗口计算并输出。允许迟到Allowed Lateness延迟关窗即使窗口第一次输出了也可以设置​allowedLatenessX​窗口在window_end​ 后再额外等 X 时间在这段时间内到来的迟到事件仍会进入窗口并触发 ​更新输出取决于输出模式/下游算子。超过 allowedLateness 的事件才会被视为“最终迟到”。Trigger / Evictor高级​Trigger自定义触发逻辑例如每来一条就触发、每 N 秒触发一次、同时满足事件数/时间等。​Evictor触发前/后剔除窗口内元素较少用成本高通常能用聚合/ProcessFunction替代。‍四、迟到数据Late Events处理当事件到达时它的事件时间戳已经 ​落后于 watermark以及超过 allowed lateness就会变成迟到数据。常见处理策略直接丢弃默认简单但会损失数据很多实时大盘能接受。侧输出Side Output收集迟到数据把迟到数据打到一个旁路流做补偿计算、落库、离线回补或告警。允许迟到并更新结果Allowed Lateness窗口结果会被修正需要下游能接收“更新/撤回”语义或你用 upsert sink。用更长乱序容忍更大的 watermark 延迟减少迟到但会增加延迟latency——典型的准确性 vs 延迟权衡。‍五、有序输出与排序如果你要的是“​按事件时间严格有序输出”Flink 不会全局帮你排序成本太高但你可以Keyed 后用KeyedProcessFunction 状态 定时器将事件暂存一段时间比如 5s等 watermark/定时器到了再按时间戳输出。代价更多状态、更高延迟、可能出现内存/状态膨胀需要 TTL、容量控制。‍六、与乱序紧密相关的运行时机制乱序本身还会影响系统行为​状态State窗口/乱序缓存都依赖 state乱序越大、allowed lateness 越大state 留存越久。​State TTL防止“永不关闭”的 key 造成状态无限增长。​Checkpoint Exactly-Once保证乱序场景下也能在故障恢复后维持一致结果尤其窗口更新、迟到补偿更依赖一致性。反压Backpressurewatermark 推进慢、窗口堆积、state 变大都可能导致下游慢→反压。‍七、常见的选型建议目标低延迟优先允许少量误差用较小乱序 watermark如 1-3s不设置/少设置 allowed lateness迟到侧输出做补偿。目标结果尽量准确延迟可接受watermark 延迟设大一些如 10-60s再配 allowed latenesssink 用 upsert/幂等写支持更新。目标必须严格按事件时间有序输出KeyedProcessFunction 缓存排序 watermark/定时器释放但要严格控制状态与延迟。‍八、代码示例8.1 DataStream API8.1.1 定义Watermark事件模型 Watermark容忍乱序publicstaticclassEvent{publicStringuserId;publicStringtype;// e.g. clickpubliclongeventTime;// epoch millispublicdoubleamount;// optionalpublicEvent(){}publicEvent(StringuserId,Stringtype,longeventTime,doubleamount){this.userIduserId;this.typetype;this.eventTimeeventTime;this.amountamount;}}WatermarkStrategy容忍 5 秒乱序 source 空闲检测避免拖慢 watermarkimportorg.apache.flink.api.common.eventtime.*;importjava.time.Duration;WatermarkStrategyEventwmWatermarkStrategy.EventforBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((e,ts)-e.eventTime).withIdleness(Duration.ofSeconds(30));// 可选分区空闲 30s 视为 idle‍8.1.2 窗口聚合// 典型大盘统计10 秒滚动窗口允许再等 10 秒 修正超过 allowed lateness 的走侧输出。importorg.apache.flink.api.common.functions.AggregateFunction;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.streaming.api.datastream.*;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;importorg.apache.flink.streaming.api.windowing.time.Time;importorg.apache.flink.util.OutputTag;publicclassWindowAggDemo{privatestaticfinalOutputTagEventLATE_TAGnewOutputTagEvent(late-events){};publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamEventsourceenv.fromElements(newEvent(u1,click,1_000L,1),newEvent(u1,click,4_000L,1),newEvent(u1,click,3_000L,1),// 乱序newEvent(u1,click,12_000L,1),newEvent(u1,click,2_000L,1)// 可能迟到);WatermarkStrategyEventwmWatermarkStrategy.EventforBoundedOutOfOrderness(java.time.Duration.ofSeconds(5)).withTimestampAssigner((e,ts)-e.eventTime).withIdleness(java.time.Duration.ofSeconds(30));SingleOutputStreamOperatorEventwithWmsource.assignTimestampsAndWatermarks(wm);SingleOutputStreamOperatorTuple2String,LongcntwithWm.keyBy(e-e.userId).window(TumblingEventTimeWindows.of(Time.seconds(10))).allowedLateness(Time.seconds(10))// 窗口结束后再等 10s.sideOutputLateData(LATE_TAG)// 超过 allowed lateness 的进侧输出.aggregate(newCountAgg());cnt.print(WINDOW);cnt.getSideOutput(LATE_TAG).print(LATE);env.execute(Window Agg Out-of-Order Demo);}publicstaticclassCountAggimplementsAggregateFunctionEvent,Long,Tuple2String,Long{privateStringkey;OverridepublicLongcreateAccumulator(){return0L;}OverridepublicLongadd(Eventv,Longacc){keyv.userId;returnacc1;}OverridepublicTuple2String,LonggetResult(Longacc){returnTuple2.of(key,acc);}OverridepublicLongmerge(Longa,Longb){returnab;}}}‍8.1.3 JoinInterval Join// 两条流都要 assign watermarks否则事件时间 join 不会按预期工作。importorg.apache.flink.streaming.api.datastream.*;importorg.apache.flink.streaming.api.functions.co.ProcessJoinFunction;importorg.apache.flink.streaming.api.windowing.time.Time;importorg.apache.flink.util.Collector;// left stream: ordersDataStreamEventorders...;// 记得 assignTimestampsAndWatermarks(wm)// right stream: paymentsDataStreamEventpayments...;// 记得 assignTimestampsAndWatermarks(wm)SingleOutputStreamOperatorStringjoinedorders.assignTimestampsAndWatermarks(wm).keyBy(e-e.userId).intervalJoin(payments.assignTimestampsAndWatermarks(wm).keyBy(e-e.userId))// 匹配条件payment.eventTime 在 order.eventTime 之后 0~600s.between(Time.seconds(0),Time.minutes(10))// 可选迟到容忍Interval Join 的迟到处理主要靠 watermark 这个参数// .withLowerBoundExclusive() / .withUpperBoundExclusive() 也可用.process(newProcessJoinFunctionEvent,Event,String(){OverridepublicvoidprocessElement(Eventleft,Eventright,Contextctx,CollectorStringout){out.collect(JOIN userleft.userId orderTsleft.eventTime payTsright.eventTime);}});joined.print(INTERVAL_JOIN);Interval Join 的关键风险状态膨胀join 本质要把一侧数据在 state 里“留一段时间等待匹配”等待窗口大between 范围大 乱序大watermark 延迟大 state 留得更久‍Window Joinimportorg.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;DataStreamEventa...;// with watermarksDataStreamEventb...;// with watermarksDataStreamStringwindowJoineda.keyBy(e-e.userId).join(b.keyBy(e-e.userId)).where(e-e.userId).equalTo(e-e.userId).window(TumblingEventTimeWindows.of(Time.seconds(10))).apply((left,right)-WIN_JOIN userleft.userId aTsleft.eventTime bTsright.eventTime);windowJoined.print(WINDOW_JOIN);‍8.2 Flink SQL8.2.1 SQL声明 watermark TUMBLE 窗口聚合CREATETABLEevents(user_id STRING,typeSTRING,tsTIMESTAMP(3),WATERMARKFORtsASts-INTERVAL5SECOND)WITH(connectorkafka,topicevents,properties.bootstrap.serverslocalhost:9092,properties.group.idg1,scan.startup.modeearliest-offset,formatjson);CREATETABLEsink_print(window_startTIMESTAMP(3),window_endTIMESTAMP(3),user_id STRING,cntBIGINT)WITH(connectorprint);INSERTINTOsink_printSELECTwindow_start,window_end,user_id,COUNT(*)AScntFROMTABLE(TUMBLE(TABLEevents,DESCRIPTOR(ts),INTERVAL10SECOND))GROUPBYwindow_start,window_end,user_id;‍8.2.2 SQL流-流 Interval Join-- 写法核心两张表都要 watermark并在 ON 条件里写清楚时间范围。CREATETABLEorders(order_id STRING,user_id STRING,tsTIMESTAMP(3),WATERMARKFORtsASts-INTERVAL5SECOND)WITH(...);CREATETABLEpayments(pay_id STRING,user_id STRING,tsTIMESTAMP(3),WATERMARKFORtsASts-INTERVAL5SECOND)WITH(...);CREATETABLEsink_join(order_id STRING,pay_id STRING,user_id STRING,order_tsTIMESTAMP(3),pay_tsTIMESTAMP(3))WITH(connectorprint);INSERTINTOsink_joinSELECTo.order_id,p.pay_id,o.user_id,o.tsASorder_ts,p.tsASpay_tsFROMorders oJOINpayments pONo.user_idp.user_id-- payment 在 order 之后 0~10分钟内ANDp.tsBETWEENo.tsANDo.tsINTERVAL10MINUTE;强烈建议SQL 里配状态 TTL防止 join 状态无限长-- 让 join / 聚合等算子的 state 有 TTL示例 2 小时你按业务改SETtable.exec.state.ttl2 h;‍