如果输入有回撤流的话，group by时间窗口出现问题

问题描述

用户在使用Flink SQL进行时间窗口聚合时遇到不支持retract stream的报错，具体场景是从业务表的CDC数据流经Kafka再到Flink进行按时间窗口的聚合。业务表实质上是一张只会产生insert操作的日志表，因此希望将retract table转换为append table以适应窗口聚合的需求。

背景信息

• Flink CDC：MongoDB CDC等插件能够捕获数据库变更数据（如INSERT、UPDATE、DELETE），并输出为 changelog stream，其中包括了-U（更新前镜像）消息，这自然形成了retract stream，即包含加（+I, +U）减（-U, -D）的消息形式。
• Retract Stream与Append Stream：Flink处理 changelog stream 时，默认生成的是retract stream，它能表达数据的增删更替，而append stream仅表示新增数据，适用于只关心数据插入的场景，如日志收集。
• Window Aggregation限制：Flink的某些窗口聚合操作直接基于append模式的表更易于实现，因为它们不直接处理数据的更新和删除。

解决方案

步骤1：识别数据特性

既然业务表是只会有INSERT操作的日志表，理论上其产生的Changelog Stream中不应包含UPDATE或DELETE操作，但通过CDC引入的数据默认会以retract stream形式存在。

步骤2：转换retract stream为append stream

要将retract stream转换为append stream，可以利用Flink的toAppendStream方法或者在SQL层面进行处理，确保只有插入操作被传递到窗口聚合操作中。

SQL层面转换

如果使用Flink SQL，可以通过创建视图来过滤掉非插入操作，从而模拟append-only的表：

CREATE VIEW append_only_log AS
SELECT * FROM cdc_log_table WHERE op = 'insert';

这里假设cdc_log_table是包含操作类型(op)字段的CDC源表，且'insert'标识插入操作。

步骤3：应用到时间窗口聚合

接下来，基于转换后的append-only视图进行时间窗口聚合操作：

SELECT window_start, window_end, COUNT(*)
FROM append_only_log
WINDOW (TUMBLE(event_time, '1 hour')) 
GROUP BY window_start, window_end;

确保聚合基于一个明确的时间戳字段（如event_time），并且使用了适合的窗口策略，如上述示例中的1小时滚动窗口。

注意事项

• 数据一致性和时效性：转换过程中需注意，虽然此方法简化了处理逻辑，但可能牺牲了处理实时更新的能力，对于需要精确反映数据变化的场景需谨慎。
• 资源消耗：直接过滤而非处理 retract 消息可能会减少状态管理的开销，但在高吞吐量场景下仍需关注整体资源消耗。

总结

针对只会产生INSERT操作的日志表，通过在Flink SQL中创建一个仅包含INSERT记录的视图，可以有效地将retract table转换为append table，进而顺利应用于时间窗口聚合场景。这种方法简化了对更新和删除操作的处理，更适合日志聚合这类只需追踪新增记录的场景。