如何持久化Apache Flink窗口最新结果 基于起止标记处理批量数据

方案1：现有窗口逻辑优化（推荐，改造成本最低） #

你现有的自定义触发器+驱逐器逻辑已经可以正确识别每一批次的完整数据，只需要补全窗口输出和Sink逻辑即可，不需要硬编码写入逻辑：

1. 窗口输出整批数据
   在现有AllWindowedStream后追加ProcessAllWindowFunction，将同窗口的所有有效数据打包为集合输出：

import com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.util.Collector;

DataStream<List<Foo>> batchResultStream = fooWindow.process(new ProcessAllWindowFunction<Foo, List<Foo>, GlobalWindow>() {
    @Override
    public void process(Context context, Iterable<Foo> elements, Collector<List<Foo>> out) {
        // 打包当前批次全量有效数据
        out.collect(Lists.newArrayList(elements));
    }
});

注意：自定义触发器触发时请返回TriggerResult.FIRE_AND_PURGE，触发计算后自动清空窗口内的旧数据，避免内存溢出。
2. 用标准Sink实现覆盖写入
你可以直接基于Flink的RichSinkFunction实现不同存储介质的覆盖逻辑，完全复用Flink的生命周期管理和故障恢复能力：

• 内存存储：在Sink的open方法中初始化ListState存储最新批次数据，invoke方法中直接更新State内容，业务侧读取State即可拿到最新数据。
• 磁盘存储：invoke方法中先将批次数据写入临时文件，写入完成后原子替换目标路径的文件，避免写入异常导致文件损坏。

方案2：CEP+Table API 实现 #

如果你想用Table API实现，不需要手动清空表，只要在Sink侧做覆盖逻辑即可：

1. 首先定义CEP的批次匹配规则，匹配Start事件->N个有效事件->End事件的序列，提取中间的有效事件得到每批次的全量数据。
2. 自定义DynamicTableSink，实现SupportsOverwrite接口，配置每次写入都为覆盖模式，每收到一个新批次的全量数据就直接替换表的存储内容即可。

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