java - 在有状态处理中，您应该如何处理由于从未达到缓冲区大小而未处理的数据？

Question

我正在全局窗口上对键进行一些聚合。仅当达到缓冲区大小时，我才会缓冲事件以进行批处理，以减少迭代次数。

但是，某些键永远不会达到缓冲区大小，因此不会被计算。

这是一些伪代码:

DoFN<> {
  @StateId("count")
  private final StateSpec<ValueState<Integer>> countState = StateSpecs.value();
  private static final int MAX_BUFFER_SIZE = 1000;
     @ProcessElement () {
       int count = firstNonNull(countState.read(), 0);
       count = count + 1;
       countState.write(count);
       if (count >= MAX_BUFFER_SIZE) {
           (...)
           countState.clear();
       }
     }
  }

效果很好，但我丢失了一些数据，因为某些键永远不会达到 1000 条记录来触发批处理。

Answer 1

我认为正确的解决方案是添加 FinishBundle方法并在 FinalizeBundle() 方法中最终确定未达到缓冲区大小的键。