Flink Interval Join后如何实现自定义聚合计算？

问题背景 #

你已通过Flink的Interval Join完成streamA（左流）与streamB（右流）的关联，代码如下：

streamA
  .keyBy((a: EventA) => a.common_key)
  .intervalJoin(
      streamB
        .keyBy((b: EventB) => b.common_key)
    )
  .between(Time.seconds(0), Time.minutes(5))
  .process(new ProcessJoinFunction<PojoA, PojoB, Result>() {
                    @Override
                    public void processElement(PojoA left, PojoB right, Context ctx, Collector<Result> out) throws Exception {
                        out.collect(Result.build(left, right));
                    }
                })

关联后生成的Result对象结构：

class Result {
   long userId; // 关联的共同Key
   String name; // 来自PojoA
   long number; // 来自PojoA
   String shop; // 来自PojoB
   long orders; // 来自PojoA
   double price; // 来自PojoA
}

当前存在单个streamA事件匹配多个streamB事件的场景，需对关联后的数据做以下聚合：

• 对orders字段求和
• 计算price的加权平均值（公式：(price1*orders1 + price2*orders2 + ...) / 总orders）
• 将聚合结果回填至Result对象

示例：

关联后记录1: (123, "nameA", 455, "shop", 3, 4.2)
关联后记录2: (123, "nameA", 455, "shop", 6, 4.8)
聚合后结果: (123, "nameA", 455, "shop", 9, 4.4)

实现方案 #

1. 定义聚合状态类 #

用于保存聚合过程中的中间累计值：

public class AggState {
    private long totalOrders;
    private double totalAmount; // 累计price*orders的值

    public AggState() {}
    public AggState(long totalOrders, double totalAmount) {
        this.totalOrders = totalOrders;
        this.totalAmount = totalAmount;
    }

    // getter & setter
    public long getTotalOrders() { return totalOrders; }
    public void setTotalOrders(long totalOrders) { this.totalOrders = totalOrders; }
    public double getTotalAmount() { return totalAmount; }
    public void setTotalAmount(double totalAmount) { this.totalAmount = totalAmount; }

    // 累加数据方法
    public void accumulate(long orders, double price) {
        this.totalOrders += orders;
        this.totalAmount += price * orders;
    }
}

2. 使用KeyedProcessFunction完成聚合 #

在Interval Join之后，按userId分组，通过KeyedProcessFunction实现聚合逻辑：

// 假设joinedStream是Interval Join输出的数据流
DataStream<Result> joinedStream = ...;

DataStream<Result> aggregatedStream = joinedStream
    .keyBy(Result::getUserId)
    .process(new KeyedProcessFunction<Long, Result, Result>() {
        // 定义状态存储聚合中间值
        private ValueState<AggState> aggState;
        // 存储维度字段（确保同一个userId的维度信息一致）
        private ValueState<Result> dimensionState;

        @Override
        public void open(Configuration parameters) throws Exception {
            // 配置状态TTL，避免状态无限膨胀
            StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.minutes(10))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();

            // 初始化聚合状态
            ValueStateDescriptor<AggState> aggDesc = new ValueStateDescriptor<>(
                "agg-state",
                AggState.class
            );
            aggDesc.enableTimeToLive(ttlConfig);
            aggState = getRuntimeContext().getState(aggDesc);

            // 初始化维度状态
            ValueStateDescriptor<Result> dimDesc = new ValueStateDescriptor<>(
                "dim-state",
                Result.class
            );
            dimDesc.enableTimeToLive(ttlConfig);
            dimensionState = getRuntimeContext().getState(dimDesc);
        }

        @Override
        public void processElement(Result value, Context ctx, Collector<Result> out) throws Exception {
            // 初始化或获取当前聚合状态
            AggState currentAgg = aggState.value();
            if (currentAgg == null) {
                currentAgg = new AggState();
            }
            // 累加数据
            currentAgg.accumulate(value.getOrders(), value.getPrice());
            aggState.update(currentAgg);

            // 保存维度信息（仅第一次处理时保存，假设同一userId维度不变）
            if (dimensionState.value() == null) {
                dimensionState.update(value);
            }

            // 计算并输出聚合结果
            Result dim = dimensionState.value();
            Result aggregatedResult = new Result();
            aggregatedResult.setUserId(dim.getUserId());
            aggregatedResult.setName(dim.getName());
            aggregatedResult.setNumber(dim.getNumber());
            aggregatedResult.setShop(dim.getShop());
            aggregatedResult.setOrders(currentAgg.getTotalOrders());
            // 避免除以0的异常
            double avgPrice = currentAgg.getTotalOrders() == 0 ? 0 : currentAgg.getTotalAmount() / currentAgg.getTotalOrders();
            aggregatedResult.setPrice(avgPrice);
            
            out.collect(aggregatedResult);
        }
    });

3. 关键注意事项 #

• 维度一致性：若同一userId对应的name/number/shop可能变化，需调整维度保存逻辑（比如取最新值）
• 状态TTL：必须配置状态过期时间，防止状态持续累积占用资源
• 异常处理：计算加权平均时需判断总订单数是否为0，避免算术异常

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