在Foundry中如何从每日快照数据集提取新增行？能否用增量转换？

关于Incremental Transform的适用性 #

完全可以用Incremental Transform来实现这个需求——它就是为这类“从全量输入中提取增量数据”的场景设计的，能高效帮你过滤出每日快照里的新增行。

核心逻辑 #

每日快照是全量数据，我们需要拿它和已经处理过的历史数据集做对比，筛选出历史中没有出现过的行。Incremental Transform的关键在于维护一个已处理数据的状态记录，用来和新快照做校验。

具体实现步骤 #

• 先确定对比标识
  必须有一个可靠的字段用来区分新旧数据：

  • 优先选业务唯一键（比如用户ID、订单ID这类主键），稳定性最高，不会因为时间、修改操作出现误差
  • 也可以用创建时间戳，但要确保这个字段是数据首次生成时的时间，不会被后续更新修改，否则会把修改过的旧数据误判为新增

• 维护状态存储
  需要一个持久化存储来记录已处理过的历史数据标识：

  • 如果用唯一键：可以建一张数据库表或者用KV存储（比如Redis），把所有已导出过的唯一键存进去
  • 如果用时间戳：只需要存储已处理过的最大创建时间戳即可，比如存在配置表或者缓存里

• 编写增量过滤逻辑
  拿到当日快照后，执行对比筛选：

  • 用唯一键的话：筛选出当日快照唯一键 NOT IN 历史唯一键集合的行，就是新增行
  • 用时间戳的话：筛选出当日快照创建时间戳 > 历史最大时间戳的行

  举个伪代码例子（SQL风格）：

  -- 用唯一键筛选新增行
  SELECT * FROM daily_snapshot
  WHERE unique_id NOT IN (SELECT unique_id FROM processed_records);
  
  -- 用时间戳筛选新增行
  SELECT * FROM daily_snapshot
  WHERE create_time > (SELECT max_create_time FROM process_status);
  

• 导出并更新状态
  把筛选出的新增行导出到目标位置后，一定要更新状态存储：

  • 用唯一键：把当日新增行的unique_id批量插入到processed_records表
  • 用时间戳：把当日快照中的最大create_time更新到process_status表

  注意：必须保证导出和状态更新的原子性，比如用事务包裹，避免导出成功但状态没更新，导致次日重复导出同一批数据。

避坑提示 #

• 如果当日快照里有重复行（同一唯一键出现多次），先对快照做去重再对比，避免重复导出
• 如果业务里有数据删除的情况，快照会去掉已删数据，但我们只关心新增，所以不用处理删除逻辑
• 状态存储要做备份，避免数据丢失导致需要重新全量处理历史数据

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