基于双Key的Apache Flink滚动加权平均实现咨询

一、核心逻辑拆解 #

先把需求拆成两步落地：

1. 按name维度单独维护每个name的最新消息（新消息直接覆盖旧消息）
2. 按parent维度聚合其下所有name的最新消息，计算加权平均（公式：sum(amount*value)/sum(amount)）

二、Key设置与状态管理细节 #

1. 单name最新消息维护 #

用KeyedStream按name做分组，借助ValueState存储每个name的完整最新消息，推荐用ProcessFunction来实现：

• 状态初始化：在open方法里定义ValueState<Message>，Message是你的数据实体类（包含parent、name、amount、value）
• 处理逻辑：每条消息进来直接覆盖对应name的状态，同时输出这条最新消息，供下一步聚合使用

2. 按parent聚合计算加权平均 #

这一步需要动态维护每个parent下所有name的最新消息集合，用KeyedStream按parent分组，搭配MapState<String, Message>存储（key是name，value是该name的最新消息）：

• 状态初始化：定义MapState来存储每个parent对应的所有name最新消息
• 处理逻辑：
  • 收到第一步传来的最新消息时，先更新MapState中对应name的条目
  • 遍历MapState里的所有消息，累加计算totalAmountValue和totalAmount，再算出加权平均
  • 注意处理totalAmount为0的异常情况，避免除以0报错

三、代码实现片段 #

1. 维护每个name的最新消息 #

public class LatestMessageByName extends KeyedProcessFunction<String, Message, Message> {
    private ValueState<Message> latestMessageState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Message> descriptor = new ValueStateDescriptor<>(
            "latestMessage",
            TypeInformation.of(Message.class)
        );
        latestMessageState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(Message msg, Context ctx, Collector<Message> out) throws Exception {
        // 直接覆盖状态，保留最新消息
        latestMessageState.update(msg);
        // 输出最新消息给下一步聚合
        out.collect(msg);
    }
}

2. 按parent聚合计算加权平均 #

public class WeightedAverageByParent extends KeyedProcessFunction<String, Message, WeightedAverageResult> {
    private MapState<String, Message> nameToLatestMsgState;

    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor<String, Message> descriptor = new MapStateDescriptor<>(
            "nameToLatestMsg",
            TypeInformation.of(String.class),
            TypeInformation.of(Message.class)
        );
        nameToLatestMsgState = getRuntimeContext().getMapState(descriptor);
    }

    @Override
    public void processElement(Message msg, Context ctx, Collector<WeightedAverageResult> out) throws Exception {
        // 更新当前parent下对应name的最新消息
        nameToLatestMsgState.put(msg.getName(), msg);

        // 计算加权平均
        double totalAmountValue = 0.0;
        double totalAmount = 0.0;
        for (Message latestMsg : nameToLatestMsgState.values()) {
            totalAmountValue += latestMsg.getAmount() * latestMsg.getValue();
            totalAmount += latestMsg.getAmount();
        }

        // 处理除数为0的情况
        double weightedAvg = totalAmount == 0 ? 0 : totalAmountValue / totalAmount;

        // 输出结果
        out.collect(new WeightedAverageResult(msg.getParent(), weightedAvg));
    }
}

3. 主流程串联 #

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 从WebSocket获取数据流并反序列化为Message对象
DataStream<Message> webSocketStream = env.socketTextStream("localhost", 8080)
    .map(new MessageDeserializer());

// 第一步：按name分组维护最新消息
DataStream<Message> latestByNameStream = webSocketStream
    .keyBy(Message::getName)
    .process(new LatestMessageByName());

// 第二步：按parent分组计算加权平均
DataStream<WeightedAverageResult> weightedAvgStream = latestByNameStream
    .keyBy(Message::getParent)
    .process(new WeightedAverageByParent());

// 输出结果
weightedAvgStream.print();

env.execute("Rolling Weighted Average Job");

四、关键注意事项 #

• 状态过期清理：如果某些name长时间无新消息，可给MapState设置TTL自动清理，避免状态无限膨胀，示例配置：

  descriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.minutes(10)).build());
  

• 滚动计算触发：这里是每来一条新消息就重新计算一次加权平均，不需要额外设置窗口，依赖状态的实时更新实现滚动计算
• 状态后端选择：如果集群数据量较大，建议使用RocksDB状态后端，支持大状态的持久化存储

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