Flink多Transformer同ID状态访问及多源输入算子实现咨询

一、跨算子访问同ID关联状态的方法 #

Flink中不同算子的状态默认相互隔离，无法直接跨算子读取，可通过以下方式实现需求：

1. 将两个流做关联（Join/CoGroup） #

把Order流和SubOrder流按OrderId关联，让SubOrder的处理逻辑直接拿到对应的Order数据，无需跨算子访问状态。示例代码：

// 定义Order流
DataStream<Order> orderStream = env.addSource(source1()).keyBy(Order::getId);
// 定义SubOrder流
DataStream<SubOrder> subOrderStream = env.addSource(source2()).keyBy(SubOrder::getOrderId);

// 使用Interval Join关联（适合有时间窗口的场景）
orderStream
    .intervalJoin(subOrderStream)
    .between(Time.minutes(-10), Time.minutes(10)) // 定义关联的时间范围
    .process(new ProcessJoinFunction<Order, SubOrder, Object>() {
        @Override
        public void processElement(Order order, SubOrder subOrder, Context ctx, Collector<Object> out) throws Exception {
            // 此处可同时获取order和subOrder，直接执行业务逻辑
            out.collect(new CombinedOrder(order, subOrder));
        }
    })
    .addSink(sink2());

2. 借助外部存储共享状态 #

将Order的状态持久化到Redis、HBase这类外部KV存储中，OrderMapper负责写入，SubOrderMapper根据OrderId查询读取。示例代码（以Redis为例）：

// OrderMapper中写入Redis
public class OrderMapper extends RichFlatMapFunction<Order, Order> {
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = new Jedis("redis-host", 6379);
    }

    @Override
    public void flatMap(Order value, Collector<Order> out) throws Exception {
        // 以orderId为key写入Redis
        jedis.set(value.getId().toString(), JSON.toJSONString(value));
        out.collect(value);
    }
}

// SubOrderMapper中读取Redis
public class SubOrderMapper extends RichFlatMapFunction<SubOrder, SubOrder> {
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = new Jedis("redis-host", 6379);
    }

    @Override
    public void flatMap(SubOrder value, Collector<SubOrder> out) throws Exception {
        // 根据orderId读取对应的Order数据
        String orderJson = jedis.get(value.getOrderId().toString());
        Order order = JSON.parseObject(orderJson, Order.class);
        // 获取到order后执行SubOrder的处理逻辑
        out.collect(value);
    }
}

3. 使用广播状态（Broadcast State） #

如果Order属于数据量小、更新不频繁的维度数据，可将Order流广播到所有SubOrderMapper实例，SubOrderMapper通过广播状态查询对应Order。示例代码：

// 定义广播状态描述符
MapStateDescriptor<String, Order> orderBroadcastStateDesc = new MapStateDescriptor<>(
    "order-broadcast-state",
    BasicTypeInfo.STRING_TYPE_INFO,
    TypeInformation.of(Order.class)
);

// 广播Order流
BroadcastStream<Order> orderBroadcastStream = orderStream.broadcast(orderBroadcastStateDesc);

// 连接SubOrder流与广播流
subOrderStream
    .connect(orderBroadcastStream)
    .process(new BroadcastProcessFunction<SubOrder, Order, Object>() {
        @Override
        public void processElement(SubOrder subOrder, ReadOnlyContext ctx, Collector<Object> out) throws Exception {
            // 从广播状态中获取对应Order
            Order order = ctx.getBroadcastState(orderBroadcastStateDesc).get(subOrder.getOrderId().toString());
            if (order != null) {
                // 执行业务处理逻辑
                out.collect(new CombinedOrder(order, subOrder));
            }
        }

        @Override
        public void processBroadcastElement(Order order, Context ctx, Collector<Object> out) throws Exception {
            // 更新广播状态中的Order数据
            ctx.getBroadcastState(orderBroadcastStateDesc).put(order.getId().toString(), order);
        }
    })
    .addSink(sink2());

二、让一个算子接收多个输入的方法 #

Flink支持多种多输入算子的实现方式，适配不同场景：

1. Union合并（同类型流） #

如果多个数据源的输出类型相同，可直接用union合并为一个流，再交给同一个算子处理：

DataStream<Event> stream1 = env.addSource(source1());
DataStream<Event> stream2 = env.addSource(source2());
DataStream<Event> stream3 = env.addSource(source3());
DataStream<Event> stream4 = env.addSource(source4());
DataStream<Event> stream5 = env.addSource(source5());

stream1.union(stream2, stream3, stream4, stream5)
    .keyBy(Event::getId)
    .flatMap(new MultiInputMapper())
    .addSink(sink());

2. ConnectedStreams连接（不同类型流） #

如果输入流类型不同，可使用connect逐步连接多个流，通过CoProcessFunction/CoFlatMapFunction处理：

DataStream<Order> orderStream = env.addSource(source1());
DataStream<SubOrder> subOrderStream = env.addSource(source2());
DataStream<Refund> refundStream = env.addSource(source3());

// 先连接两个流，再连接第三个流
ConnectedStreams<Order, SubOrder> connected1 = orderStream.connect(subOrderStream);
ConnectedStreams<ConnectedStreams<Order, SubOrder>, Refund> connectedAll = connected1.connect(refundStream);

// 自定义CoProcessFunction处理多类型输入
connectedAll.process(new CustomCoProcessFunction()).addSink(sink());

3. MultipleInputOperator（Flink 1.13+，多异构流） #

专门用于处理多个不同类型输入流的算子，每个输入流对应独立的处理函数，适合5个不同类型数据源的场景：

// 定义每个输入流的处理函数
class OrderInputFunction implements InputFormatFunction<Order, Object> {
    @Override
    public void processElement(Order value, Context ctx, Collector<Object> out) throws Exception {
        // 处理Order类型数据
    }
}

class SubOrderInputFunction implements InputFormatFunction<SubOrder, Object> {
    @Override
    public void processElement(SubOrder value, Context ctx, Collector<Object> out) throws Exception {
        // 处理SubOrder类型数据
    }
}

// 构建多输入算子
env.fromMultipleInputs(
    InputConfig.of(env.addSource(source1()), new OrderInputFunction()),
    InputConfig.of(env.addSource(source2()), new SubOrderInputFunction()),
    InputConfig.of(env.addSource(source3()), new RefundInputFunction()),
    InputConfig.of(env.addSource(source4()), new LogInputFunction()),
    InputConfig.of(env.addSource(source5()), new MetricInputFunction())
).addSink(sink());

内容的提问来源于stack exchange，提问作者Abidi