Flink DataStream API多流操作与实践

一、概览多流操作 DataStream API ├─ union() 同类型多流合并 ├─ connect() 两条不同类型流连接 │ ├─ CoMapFunction / CoFlatMapFunction │ ├─ CoProcessFunction │ └─ KeyedCoProcessFunction ├─ broadcast() connect() 规则流 / 配置流广播 │ ├─ BroadcastProcessFunction │ └─ KeyedBroadcastProcessFunction ├─ intervalJoin() KeyedStream 双流时间区间 Join ├─ join() 窗口 Join ├─ coGroup() 窗口 CoGroup类似全量分组关联 └─ side output 主流拆出多条旁路流常用于告警/脏数据二、union()同类型多流合并适合多个来源结构完全一致的流例如多个 Kafka topic、多个业务线的订单流。DataStreamOrder appOrders env.fromSource(appSource, wm, app-orders); DataStreamOrder webOrders env.fromSource(webSource, wm, web-orders); DataStreamOrder posOrders env.fromSource(posSource, wm, pos-orders); DataStreamOrder allOrders appOrders .union(webOrders, posOrders);特点只能合并相同数据类型的流。不做去重、不做排序只是把多条流合成一条流。多流 watermark 会取各输入流 watermark 的最小值因此某一路延迟会拖慢整体事件时间推进。最佳实践多源合并前尽量统一 schema。如果多个源可能重复生产同一事件union() 后要基于业务主键做去重。对空闲 source 配置 idleness避免某个分区无数据导致 watermark 卡住。WatermarkStrategyOrder wm WatermarkStrategy .OrderforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((e, ts) - e.getEventTime()) .withIdleness(Duration.ofMinutes(1));三、connect()两条不同类型流连接处理connect()可以连接两条不同类型的流得到ConnectedStreams再用双输入函数处理。CoMapFunction/CoFlatMapFunction适合简单双流转换。DataStreamOrder orders ...; DataStreamUserEvent userEvents ...; DataStreamString result orders .connect(userEvents) .flatMap(new CoFlatMapFunctionOrder, UserEvent, String() { Override public void flatMap1(Order order, CollectorString out) { out.collect(order: order.getOrderId()); } Override public void flatMap2(UserEvent event, CollectorString out) { out.collect(event: event.getUserId()); } });其中flatMap1()处理第一条流flatMap2()处理第二条流两条流类型可以不同但输出类型相同。CoProcessFunction适合需要状态、定时器、复杂逻辑的双流处理例如订单流关联支付流订单先到则缓存支付先到也缓存超时则输出未支付订单。DataStreamOrder orders ...; DataStreamPayment payments ...; DataStreamResult result orders .keyBy(Order::getOrderId) .connect(payments.keyBy(Payment::getOrderId)) .process(new KeyedCoProcessFunctionString, Order, Payment, Result() { private ValueStateOrder orderState; private ValueStatePayment paymentState; Override public void open(Configuration parameters) { orderState getRuntimeContext().getState( new ValueStateDescriptor(orderState, Order.class)); paymentState getRuntimeContext().getState( new ValueStateDescriptor(paymentState, Payment.class)); } Override public void processElement1( Order order, Context ctx, CollectorResult out) throws Exception { Payment payment paymentState.value(); if (payment ! null) { out.collect(Result.matched(order, payment)); paymentState.clear(); } else { orderState.update(order); ctx.timerService().registerEventTimeTimer( order.getEventTime() 15 * 60 * 1000L); } } Override public void processElement2( Payment payment, Context ctx, CollectorResult out) throws Exception { Order order orderState.value(); if (order ! null) { out.collect(Result.matched(order, payment)); orderState.clear(); } else { paymentState.update(payment); } } Override public void onTimer( long timestamp, OnTimerContext ctx, CollectorResult out) throws Exception { Order order orderState.value(); if (order ! null) { out.collect(Result.timeout(order)); orderState.clear(); } } });最佳实践双流关联优先使用 keyBy(...).connect(...).process(...)。两条流必须按相同业务 key 分区否则同一个订单和支付可能到不同 Task。状态一定要清理避免长期膨胀。有超时语义时优先使用事件时间定时器。状态较大时开启 RocksDB StateBackend 和 TTL。四、broadcast()广播流关联主流适合“配置流 / 规则流 / 维表变更流”广播到所有并行实例。典型场景风控规则实时更新。活动配置实时生效。黑名单、白名单、阈值规则更新。MapStateDescriptorString, Rule ruleStateDesc new MapStateDescriptor(rules, String.class, Rule.class); BroadcastStreamRule ruleBroadcastStream rules.broadcast(ruleStateDesc); DataStreamAlert alerts events .keyBy(Event::getUserId) .connect(ruleBroadcastStream) .process(new KeyedBroadcastProcessFunctionString, Event, Rule, Alert() { Override public void processElement( Event event, ReadOnlyContext ctx, CollectorAlert out) throws Exception { ReadOnlyBroadcastStateString, Rule rules ctx.getBroadcastState(ruleStateDesc); for (Map.EntryString, Rule entry : rules.immutableEntries()) { Rule rule entry.getValue(); if (rule.match(event)) { out.collect(new Alert(event, rule)); } } } Override public void processBroadcastElement( Rule rule, Context ctx, CollectorAlert out) throws Exception { BroadcastStateString, Rule state ctx.getBroadcastState(ruleStateDesc); state.put(rule.getRuleId(), rule); } });最佳实践广播流数据量不能太大因为每个并行实例都会保存一份。规则要有版本号、更新时间、启停状态。支持删除规则不要只支持新增/覆盖。广播状态适合小规模动态配置不适合大维表。大维表优先考虑异步 IO、外部 KV、Lookup Join 或 Table/SQL。五、intervalJoin()双流时间区间 Join适合两条 keyed 流按事件时间做区间关联。例如订单事件和支付事件支付时间在订单后 15 分钟内即认为匹配。DataStreamOrderPayment result orders .keyBy(Order::getOrderId) .intervalJoin(payments.keyBy(Payment::getOrderId)) .between(Time.minutes(0), Time.minutes(15)) .process(new ProcessJoinFunctionOrder, Payment, OrderPayment() { Override public void processElement( Order order, Payment payment, Context ctx, CollectorOrderPayment out) { out.collect(new OrderPayment(order, payment)); } });特点只能用于 KeyedStream。基于事件时间。会根据时间区间缓存两边数据。不匹配的数据默认不会输出。最佳实践用于明确的时间窗口关联比如“点击后 10 分钟内下单”。时间区间不要过大否则状态会膨胀。要求两条流都有正确 watermark。如果需要输出未匹配数据intervalJoin() 不够灵活可以用 KeyedCoProcessFunction 自己实现。六、join()窗口 Join适合按窗口做内连接。DataStreamOrderPayment result orders .join(payments) .where(Order::getOrderId) .equalTo(Payment::getOrderId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .apply(new JoinFunctionOrder, Payment, OrderPayment() { Override public OrderPayment join(Order order, Payment payment) { return new OrderPayment(order, payment); } });特点两边在同一个窗口内且 key 相等时才会 join。本质是窗口内笛卡尔匹配。没有匹配的数据不会输出。最佳实践适合批式窗口关联例如每 5 分钟内订单和支付匹配。如果关联语义是“订单后 15 分钟内支付”优先用 intervalJoin()。窗口 join 对迟到数据敏感要配置 watermark 和 allowed lateness。七、coGroup()窗口内双流分组处理coGroup()比join()更灵活可以拿到窗口内两边的全部元素。DataStreamResult result orders .coGroup(payments) .where(Order::getOrderId) .equalTo(Payment::getOrderId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .apply(new CoGroupFunctionOrder, Payment, Result() { Override public void coGroup( IterableOrder orderGroup, IterablePayment paymentGroup, CollectorResult out) { ListOrder orderList Lists.newArrayList(orderGroup); ListPayment paymentList Lists.newArrayList(paymentGroup); if (paymentList.isEmpty()) { for (Order order : orderList) { out.collect(Result.unpaid(order)); } } else { for (Order order : orderList) { for (Payment payment : paymentList) { out.collect(Result.matched(order, payment)); } } } } });适合需要处理未匹配数据。需要一对多、多对多关联。需要自定义窗口内关联逻辑。八、Side Output一条流拆出多条旁路流虽然不是典型“多输入流”但常用于多流处理架构中把主流拆成告警流、脏数据流、迟到数据流等。OutputTagDirtyRecord dirtyTag new OutputTagDirtyRecord(dirty) {}; SingleOutputStreamOperatorEvent cleanStream rawStream .process(new ProcessFunctionRawEvent, Event() { Override public void processElement( RawEvent value, Context ctx, CollectorEvent out) { if (value.isValid()) { out.collect(value.toEvent()); } else { ctx.output(dirtyTag, new DirtyRecord(value)); } } }); DataStreamDirtyRecord dirtyStream cleanStream.getSideOutput(dirtyTag);最佳实践脏数据、异常数据不要直接丢弃优先通过 side output 输出到旁路。迟到数据可以输出到独立 Kafka topic 供补偿处理。主流程和异常流程解耦便于监控和回放。九、选型建议场景推荐 API多条同类型流合并union()两条不同类型流简单处理connect() CoMapFunction / CoFlatMapFunction双流复杂状态关联connect() KeyedCoProcessFunction主流关联动态规则流broadcast() KeyedBroadcastProcessFunction两条 keyed 流按时间范围关联intervalJoin()两条流按窗口内 key 关联join()需要处理未匹配或复杂窗口分组coGroup()主流拆告警/脏数据/迟到数据side output