Apache Flink并行实例算子状态分发与键控流处理结果不符的疑问

你的问题本质是算子状态的分片分发和键控流的分区路由共同导致的结果不符合预期：每个并行实例只拿到部分乘法器，而键控流又把相同键的元素固定路由到同一个实例，最终每个元素只能被部分乘数处理。要实现所有元素被全部乘法器处理，同时保持并行度，有两种可行方案，取决于你的乘法器列表是静态还是动态的：

方案1：静态乘法器列表 - 每个实例持有完整状态 #

如果你的乘法器列表是固定不变的（比如示例中的[2,3,4,5]），最简单的方式是让每个并行算子实例都初始化完整的乘法器列表，不需要依赖Flink的状态分发机制。这样不管元素被路由到哪个实例，都能拿到全部乘数进行计算。

修改后的代码示例：

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.util.Arrays;
import java.util.List;

class MultiplyNumber implements FlatMapFunction<Integer, Integer> {
    private List<Integer> multipliers;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 每个实例都初始化完整的乘法器列表
        multipliers = Arrays.asList(2, 3, 4, 5);
    }

    @Override
    public void flatMap(Integer value, Collector<Integer> out) {
        for (Integer multiplier : multipliers) {
            out.collect(multiplier * value);
        }
    }
}

这种方案的优势是简单高效，不需要额外的状态管理开销，适合静态不变的配置类状态。

方案2：动态乘法器列表 - 使用Broadcast State（广播状态） #

如果乘法器列表需要动态更新（比如从外部配置中心或数据流中获取更新），那么需要使用Flink的广播状态。广播状态会将状态完整地广播到所有并行算子实例，确保每个实例都持有全量的状态副本，这样无论元素被路由到哪个实例，都能使用最新的全量乘法器进行计算。

实现步骤： #

1. 定义广播状态的描述符，用于标识广播状态的类型和名称；
2. 创建广播流，将乘法器的更新数据广播到所有并行实例；
3. 将主流（你的键控整数流）与广播流连接，在BroadcastProcessFunction中处理每个元素时使用广播状态中的全量乘法器。

代码示例：

import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.ListTypeInfo;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import java.util.Arrays;
import java.util.List;

public class BroadcastStateExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // 1. 定义广播状态描述符
        MapStateDescriptor<String, List<Integer>> multiplierStateDescriptor =
                new MapStateDescriptor<>(
                        "multipliers-state",
                        BasicTypeInfo.STRING_TYPE_INFO,
                        new ListTypeInfo<>(Integer.class)
                );

        // 2. 创建广播流（这里用静态示例数据，实际可以替换为动态更新的数据源）
        DataStream<List<Integer>> broadcastSource = env.fromElements(Arrays.asList(2, 3, 4, 5));
        BroadcastStream<List<Integer>> broadcastStream = broadcastSource.broadcast(multiplierStateDescriptor);

        // 3. 你的主流（键控整数流）
        DataStream<Integer> mainStream = env.fromElements(1, 2, 3, 4, 5); // 示例输入流

        // 4. 连接主流与广播流，处理每个元素
        mainStream.connect(broadcastStream)
                .process(new BroadcastProcessFunction<Integer, List<Integer>, Integer>() {
                    @Override
                    public void processElement(Integer value, ReadOnlyContext ctx, Collector<Integer> out) throws Exception {
                        // 获取广播状态中的全量乘法器列表
                        List<Integer> multipliers = ctx.getBroadcastState(multiplierStateDescriptor).get("multipliers");
                        if (multipliers != null) {
                            for (Integer multiplier : multipliers) {
                                out.collect(multiplier * value);
                            }
                        }
                    }

                    @Override
                    public void processBroadcastElement(List<Integer> multipliers, Context ctx, Collector<Integer> out) throws Exception {
                        // 更新广播状态，所有并行实例都会同步这个更新
                        ctx.getBroadcastState(multiplierStateDescriptor).put("multipliers", multipliers);
                    }
                })
                .print();

        env.execute("Broadcast State Example");
    }
}

为什么原来的方式不符合预期？ #

你最初的代码中，算子状态被Flink默认分片分发（比如ListState会将列表拆分为多个子列表分配给不同并行实例），再加上键控流的分区策略（相同键的元素固定路由到同一个实例），导致偶数元素只能被Operator1的[2,3]处理，奇数元素只能被Operator2的[4,5]处理，最终无法得到全量的计算结果。而上面的两种方案都确保了每个并行实例能拿到完整的乘法器列表，从而让每个输入元素都能被所有乘数处理。

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