Flink ProcessFunction每5秒输出异常：如何仅输出正确分组求和值？

问题原因 #

当前代码中每处理一个元素就注册一个5秒后的ProcessingTime定时器，导致同一key下存在多个定时器。第一个定时器触发时，求和状态已累计到100，输出后清空了状态；后续定时器触发时，状态已为空，因此输出null。

修改方案 #

每次处理新元素时，先取消之前注册的旧定时器，仅保留最新的5秒后定时器。这样能确保只有在连续5秒没有该key的新元素时，才触发onTimer输出最终求和结果。

修改后的代码 #

public static class MyKeyedProcessFunction
      extends KeyedProcessFunction<String, Tuple2<String, Integer>, Tuple2<String, Integer>> {
    private ValueState<Integer> sumState;
    private ValueState<Long> timerState; // 存储已注册的定时器时间戳

    @Override
    public void open(Configuration parameters) throws Exception {
      sumState = getRuntimeContext().getState(new ValueStateDescriptor<>("sum", Integer.class));
      timerState = getRuntimeContext().getState(new ValueStateDescriptor<>("timer", Long.class));
    }

    @Override
    public void processElement(
        Tuple2<String, Integer> value,
        Context ctx,
        Collector<Tuple2<String, Integer>> out) throws Exception {
      // 累计求和
      Integer sum = sumState.value();
      sum = sum == null ? 0 : sum;
      sum += value.f1;
      sumState.update(sum);

      // 取消旧定时器（如果存在）
      Long oldTimer = timerState.value();
      if (oldTimer != null) {
        ctx.timerService().deleteProcessingTimeTimer(oldTimer);
      }

      // 注册新的5秒后定时器
      long newTimer = ctx.timerService().currentProcessingTime() + 5000;
      ctx.timerService().registerProcessingTimeTimer(newTimer);
      timerState.update(newTimer);
    }

    @Override
    public void onTimer(
        long timestamp,
        OnTimerContext ctx,
        Collector<Tuple2<String, Integer>> out) throws Exception {
      // 输出最终求和结果
      out.collect(Tuple2.of(ctx.getCurrentKey(), sumState.value()));
      // 清空状态
      sumState.clear();
      timerState.clear();
    }
  }

关键修改点 #

• 新增timerState存储当前key下已注册的定时器时间戳，用于后续取消旧定时器
• 在processElement中，每次处理新元素时先删除旧定时器，避免多个定时器同时存在
• 仅保留最新的5秒后定时器，确保只有在无新元素输入的5秒后，才输出最终求和结果

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