使用ROW_NUMBER获取客户最新记录时遇Table sink不支持更新删除变更错误

问题根源 #

流处理模式下，ROW_NUMBER()这类Rank算子会生成Upsert类型数据流：当客户产生新的最新记录时，旧的"最新记录"会被标记为删除事件，新记录作为插入/更新事件输出。若你的输出sink仅支持追加模式（如普通文件输出、未配置Upsert的Kafka Sink），就会触发该报错。

可行解决方案 #

根据业务场景选择对应方案：

方案1：强制批处理模式（离线静态数据场景） #

如果处理的是离线静态表，在SQL开头添加批处理优化器提示，让Rank算子仅生成最终结果集，不产生更新/删除事件：

/*+ OPTIMIZER_MODE(BATCH) */
SELECT CUST_ID, DEPT_NAME, CREATED_AT 
FROM (
  SELECT *, ROW_NUMBER() OVER(PARTITION BY CUST_ID ORDER BY CREATED_AT DESC) AS row_num
  FROM inputTable
)
WHERE row_num = 1;

方案2：分组取最大时间戳关联（流处理+仅支持追加sink场景） #

若需实时处理流数据且sink仅支持追加模式，可替换Rank算子为分组取最大时间戳后关联原表的写法，生成纯追加流：

SELECT t.CUST_ID, t.DEPT_NAME, t.CREATED_AT
FROM inputTable t
INNER JOIN (
  SELECT CUST_ID, MAX(CREATED_AT) AS latest_created_at
  FROM inputTable
  GROUP BY CUST_ID
) latest ON t.CUST_ID = latest.CUST_ID AND t.CREATED_AT = latest.latest_created_at;

注意：若同一客户同一时间存在多条记录，该语句会返回所有符合记录；需确保CREATED_AT在客户维度唯一才能得到单条最新记录。

方案3：使用支持Upsert模式的Sink（实时更新场景） #

若需实时维护每个客户的最新记录（旧记录需被替换），需将sink配置为支持Upsert模式：

• Flink JDBC Sink：配置sink.upsert-mode = 'upsert'，并将CUST_ID设为主键
• Upsert Kafka Sink：指定主键字段，Kafka会自动处理更新/删除事件
  配置完成后，原Rank语句即可正常执行。

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