Flink SQL使用toRetractStream做表关联时如何获取最终正确计算结果

你目前使用的是Flink的常规流Join，这类Join默认没有配置输出触发时机，采用事件驱动的输出逻辑：

• 当A表数据先到达Join算子时，B表暂无匹配数据，会先输出一条(true, 仅A表计算结果)的正向记录
• 当B表匹配数据后续到达时，先输出一条(false, 仅A表计算结果)的撤回记录删除之前的错误结果，再输出(true, A+B关联后的正确结果)的正向记录
  这就是你看到3条记录的核心原因。

方案1：使用窗口Join（推荐，适配你的窗口聚合场景） #

你的两张表都是窗口聚合后的结果，都携带window_start_time和window_end_time字段，可通过配置水位线+窗口关联规则，让Flink感知窗口结束时机，仅在窗口数据全部到齐后输出最终结果，不会产生中间撤回记录。

1. 首先给两个输入流定义事件时间水位线：

// 以window_end_time作为事件时间，根据业务需要调整水位线延迟时长
streamA = streamA.assignTimestampsAndWatermarks(WatermarkStrategy
        .<Row>forMonotonousTimestamps()
        .withTimestampAssigner((row, ts) -> row.getFieldAs("window_end_time"))
);
streamB = streamB.assignTimestampsAndWatermarks(WatermarkStrategy
        .<Row>forMonotonousTimestamps()
        .withTimestampAssigner((row, ts) -> row.getFieldAs("window_end_time"))
);

2. 改写关联SQL，增加窗口结束触发条件：

select 
  A.speed_sum + COALESCE(B.speed_sum,0.0) as total_speed,
  A.cnt + COALESCE(B.cnt,0) as total_cnt,
  A.window_start_time, 
  A.window_end_time 
from A 
left join B 
on 
  A.window_start_time = B.window_start_time
  and A.window_end_time = B.window_end_time
  and B.window_end_time >= A.window_end_time

配置后输出结果仅包含最终正确记录，无多余撤回数据。

方案2：DataStream层结果去重（临时适配方案） #

如果不需要修改原有SQL逻辑，也可以基于窗口唯一键（window_start_time+window_end_time）做状态去重，仅保留每个窗口的最终有效结果：

DataStream<Row> finalResult = streamResult
  // 按窗口维度分组
  .keyBy(t -> Tuple2.of(t.f1.getField("window_start_time"), t.f1.getField("window_end_time")))
  .process(new KeyedProcessFunction<Tuple2<Long, Long>, Tuple2<Boolean, Row>, Row>() {
    private ValueState<Row> latestResultState;
    @Override
    public void open(Configuration parameters) throws Exception {
      latestResultState = getRuntimeContext().getState(
        new ValueStateDescriptor<>("latestResult", Row.class)
      );
    }
    @Override
    public void processElement(Tuple2<Boolean, Row> value, Context ctx, Collector<Row> out) throws Exception {
      if (value.f0) {
        // 保存最新的正向结果
        latestResultState.update(value.f1);
        // 注册窗口结束定时器，到点输出结果
        long windowEnd = value.f1.getFieldAs("window_end_time");
        ctx.timerService().registerEventTimeTimer(windowEnd);
      } else {
        // 撤回时清空旧状态
        latestResultState.clear();
      }
    }
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Row> out) throws Exception {
      if (latestResultState.value() != null) {
        out.collect(latestResultState.value());
        latestResultState.clear();
      }
    }
  });
finalResult.print("finalResult");

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