如何在Flink SQL分组窗口聚合中删除MongoDB旧状态？

首先纠正你当前SQL的问题：你的GROUP BY子句用了GROUPING SETS ((PURCHASE_TIMESTAMP))，这会导致聚合维度是窗口+购买时间，而非你示例中的用户维度（USER_ID）。正确的窗口分组聚合SQL应该是：

CREATE VIEW USER_TABLE
AS
SELECT
    window_start AS WINDOW_START,
    window_end AS WINDOW_END,
    USER_ID,
    SUM(PURCHASE_AMOUNT) AS PURCHASE_AMOUNT,
    COUNT(*) AS PURCHASE_COUNT,
    MAX(PURCHASE_TIMESTAMP) AS LAST_ACTIVE_TIME -- 新增最后活跃时间字段
FROM TABLE(
    HOP(
      DATA => TABLE USER_SRC,
      TIMECOL => DESCRIPTOR(PURCHASE_TIMESTAMP),
      SLIDE => INTERVAL '1' DAY,
      SIZE => INTERVAL '5' DAYS))
GROUP BY window_start, window_end, USER_ID;

回到你的核心需求：窗口内无购买行为的用户，需要删除其MongoDB中的旧记录或更新为0。问题根源在于Flink窗口聚合仅会输出有数据更新的Key，无数据的用户不会生成聚合结果，因此无法触发MongoDB的更新/删除操作。以下是三种可行的解决方案：

方案1：MongoDB TTL索引自动清理（最简单） #

利用MongoDB的TTL索引特性，自动删除超过指定时间未更新的用户记录：

1. 修改Flink聚合逻辑：在输出字段中添加LAST_ACTIVE_TIME（可以取窗口的window_end或用户最新的PURCHASE_TIMESTAMP），确保每次用户有购买行为时，该字段会被更新为最新时间。
2. 在MongoDB中创建TTL索引：

   -- 为LAST_ACTIVE_TIME字段设置30天过期（2592000秒）
   db.user_purchases.createIndex({LAST_ACTIVE_TIME: 1}, {expireAfterSeconds: 2592000})
   

3. 效果：用户近30天无购买行为时，LAST_ACTIVE_TIME不会被更新，到期后MongoDB自动删除该文档。完全不需要修改Flink的Sink逻辑，适合大多数场景。

方案2：补全无数据用户的聚合结果（输出0值） #

如果需要在无数据窗口中主动输出0值而非删除，可以通过关联全量用户维度表补全所有用户的聚合结果：

1. 准备全量用户维度表：假设你有一个存储所有用户ID的表ALL_USERS（可以从MySQL/MongoDB等数据源同步）。
2. 关联聚合结果与维度表：

   CREATE VIEW FULL_USER_TABLE
   AS
   SELECT
       w.WINDOW_START,
       w.WINDOW_END,
       u.USER_ID,
       COALESCE(w.PURCHASE_AMOUNT, 0) AS PURCHASE_AMOUNT,
       COALESCE(w.PURCHASE_COUNT, 0) AS PURCHASE_COUNT,
       CURRENT_TIMESTAMP() AS LAST_ACTIVE_TIME -- 无数据时更新为当前窗口时间
   FROM ALL_USERS u
   LEFT JOIN USER_TABLE w
       ON u.USER_ID = w.USER_ID
       AND w.WINDOW_START = CURRENT_WINDOW_START() -- 根据滑动窗口逻辑匹配当前窗口
   

3. 写入MongoDB：将FULL_USER_TABLE以Upsert模式写入MongoDB（主键设为USER_ID + WINDOW_START），这样无数据用户的记录会被更新为0值。

方案3：利用Flink状态TTL生成DELETE事件 #

如果需要主动触发MongoDB的DELETE操作，可以结合Flink的状态TTL和自定义Sink实现：

1. 调整聚合逻辑为用户级状态：放弃窗口级聚合，改为基于用户的无界聚合，跟踪用户的最后活跃时间和累计购买数据。
2. 设置状态TTL：在Flink SQL中通过STATE_TTL配置状态过期时间（比如30天）：

   CREATE TABLE USER_AGG(
       USER_ID STRING PRIMARY KEY,
       PURCHASE_AMOUNT BIGINT,
       PURCHASE_COUNT BIGINT,
       LAST_ACTIVE_TIME TIMESTAMP_LTZ(3)
   ) WITH (
       'connector' = 'mongodb',
       'database' = 'test',
       'collection' = 'user_purchases',
       'state.ttl' = '864000000' -- 30天（单位：毫秒）
   );
   

3. 自定义Sink处理过期状态：当用户状态过期时，Flink会触发状态清理，此时自定义Sink可以捕获该事件并向MongoDB发送DELETE指令。

针对你的示例场景的适配 #

以方案1为例，你的示例流程会变成：

• Window 1：Key1和Key2的LAST_ACTIVE_TIME被设为窗口1的end时间，写入MongoDB。
• Window 2：Key2的LAST_ACTIVE_TIME更新为窗口2的end时间；Key1无数据，LAST_ACTIVE_TIME保持窗口1的时间。
• 30天后：Key1的LAST_ACTIVE_TIME过期，MongoDB自动删除该记录；Key2的记录保留。
• Window 3：Key1的LAST_ACTIVE_TIME更新为窗口3的end时间，记录被重新写入MongoDB。

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