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、多个业务线的订单流。

DataStream<Order> appOrders = env.fromSource(appSource, wm, "app-orders"); DataStream<Order> webOrders = env.fromSource(webSource, wm, "web-orders"); DataStream<Order> posOrders = env.fromSource(posSource, wm, "pos-orders"); DataStream<Order> allOrders = appOrders .union(webOrders, posOrders);

特点：

• 只能合并相同数据类型的流。
• 不做去重、不做排序，只是把多条流合成一条流。
• 多流 watermark 会取各输入流 watermark 的最小值，因此某一路延迟会拖慢整体事件时间推进。

最佳实践：

• 多源合并前尽量统一 schema。
• 如果多个源可能重复生产同一事件，union() 后要基于业务主键做去重。
• 对空闲 source 配置 idleness，避免某个分区无数据导致 watermark 卡住。

WatermarkStrategy<Order> wm = WatermarkStrategy .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((e, ts) -> e.getEventTime()) .withIdleness(Duration.ofMinutes(1));

三、connect()：两条不同类型流连接处理

connect()可以连接两条不同类型的流，得到ConnectedStreams，再用双输入函数处理。

CoMapFunction/CoFlatMapFunction适合简单双流转换。

DataStream<Order> orders = ...; DataStream<UserEvent> userEvents = ...; DataStream<String> result = orders .connect(userEvents) .flatMap(new CoFlatMapFunction<Order, UserEvent, String>() { @Override public void flatMap1(Order order, Collector<String> out) { out.collect("order: " + order.getOrderId()); } @Override public void flatMap2(UserEvent event, Collector<String> out) { out.collect("event: " + event.getUserId()); } });

其中flatMap1()处理第一条流，flatMap2()处理第二条流，两条流类型可以不同，但输出类型相同。

CoProcessFunction适合需要状态、定时器、复杂逻辑的双流处理，例如订单流关联支付流，订单先到则缓存，支付先到也缓存，超时则输出未支付订单。

DataStream<Order> orders = ...; DataStream<Payment> payments = ...; DataStream<Result> result = orders .keyBy(Order::getOrderId) .connect(payments.keyBy(Payment::getOrderId)) .process(new KeyedCoProcessFunction<String, Order, Payment, Result>() { private ValueState<Order> orderState; private ValueState<Payment> 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, Collector<Result> 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, Collector<Result> 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, Collector<Result> 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()：广播流关联主流

适合“配置流 / 规则流 / 维表变更流”广播到所有并行实例。

典型场景：

• 风控规则实时更新。
• 活动配置实时生效。
• 黑名单、白名单、阈值规则更新。

MapStateDescriptor<String, Rule> ruleStateDesc = new MapStateDescriptor<>("rules", String.class, Rule.class); BroadcastStream<Rule> ruleBroadcastStream = rules.broadcast(ruleStateDesc); DataStream<Alert> alerts = events .keyBy(Event::getUserId) .connect(ruleBroadcastStream) .process(new KeyedBroadcastProcessFunction<String, Event, Rule, Alert>() { @Override public void processElement( Event event, ReadOnlyContext ctx, Collector<Alert> out) throws Exception { ReadOnlyBroadcastState<String, Rule> rules = ctx.getBroadcastState(ruleStateDesc); for (Map.Entry<String, 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, Collector<Alert> out) throws Exception { BroadcastState<String, Rule> state = ctx.getBroadcastState(ruleStateDesc); state.put(rule.getRuleId(), rule); } });

最佳实践：

• 广播流数据量不能太大，因为每个并行实例都会保存一份。
• 规则要有版本号、更新时间、启停状态。
• 支持删除规则，不要只支持新增/覆盖。
• 广播状态适合小规模动态配置，不适合大维表。
• 大维表优先考虑异步 IO、外部 KV、Lookup Join 或 Table/SQL。

五、intervalJoin()：双流时间区间 Join

适合两条 keyed 流按事件时间做区间关联。

例如：订单事件和支付事件，支付时间在订单后 15 分钟内即认为匹配。

DataStream<OrderPayment> result = orders .keyBy(Order::getOrderId) .intervalJoin(payments.keyBy(Payment::getOrderId)) .between(Time.minutes(0), Time.minutes(15)) .process(new ProcessJoinFunction<Order, Payment, OrderPayment>() { @Override public void processElement( Order order, Payment payment, Context ctx, Collector<OrderPayment> out) { out.collect(new OrderPayment(order, payment)); } });

特点：

• 只能用于 KeyedStream。
• 基于事件时间。
• 会根据时间区间缓存两边数据。
• 不匹配的数据默认不会输出。

最佳实践：

• 用于明确的时间窗口关联，比如“点击后 10 分钟内下单”。
• 时间区间不要过大，否则状态会膨胀。
• 要求两条流都有正确 watermark。
• 如果需要输出未匹配数据，intervalJoin() 不够灵活，可以用 KeyedCoProcessFunction 自己实现。

六、join()：窗口 Join

适合按窗口做内连接。

DataStream<OrderPayment> result = orders .join(payments) .where(Order::getOrderId) .equalTo(Payment::getOrderId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .apply(new JoinFunction<Order, 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()更灵活，可以拿到窗口内两边的全部元素。

DataStream<Result> result = orders .coGroup(payments) .where(Order::getOrderId) .equalTo(Payment::getOrderId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .apply(new CoGroupFunction<Order, Payment, Result>() { @Override public void coGroup( Iterable<Order> orderGroup, Iterable<Payment> paymentGroup, Collector<Result> out) { List<Order> orderList = Lists.newArrayList(orderGroup); List<Payment> 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：一条流拆出多条旁路流

虽然不是典型“多输入流”，但常用于多流处理架构中，把主流拆成告警流、脏数据流、迟到数据流等。

OutputTag<DirtyRecord> dirtyTag = new OutputTag<DirtyRecord>("dirty") {}; SingleOutputStreamOperator<Event> cleanStream = rawStream .process(new ProcessFunction<RawEvent, Event>() { @Override public void processElement( RawEvent value, Context ctx, Collector<Event> out) { if (value.isValid()) { out.collect(value.toEvent()); } else { ctx.output(dirtyTag, new DirtyRecord(value)); } } }); DataStream<DirtyRecord> dirtyStream = cleanStream.getSideOutput(dirtyTag);

最佳实践：

• 脏数据、异常数据不要直接丢弃，优先通过 side output 输出到旁路。
• 迟到数据可以输出到独立 Kafka topic 供补偿处理。
• 主流程和异常流程解耦，便于监控和回放。

九、选型建议