Snowflake Stream数据异常丢失求助：部分日期未处理却被清空

Snowflake Stream的核心特性是：只要通过SELECT查询了Stream的变更记录（元数据查询除外），这些记录就会被标记为已消费，后续无法再访问。你的游标在遍历所有process_date时，会扫描到那些仅在2个Stream中有数据的日期，哪怕你没对这些数据执行存储过程，对应的Stream条目已经被视为已处理，导致数据丢失。

1. 利用事务+Stream的事务级消费特性（原生特性，推荐） #

Snowflake Stream的消费标记是绑定到事务的：在一个事务内查询Stream的数据，只有当事务提交时，这些被查询到的数据才会被标记为已消费。基于这个特性调整逻辑：

• 开启事务，在事务内完成以下操作：
  1. 找出三个Stream中共同存在的process_date：

     WITH common_dates AS (
         SELECT process_date FROM stream1
         INTERSECT
         SELECT process_date FROM stream2
         INTERSECT
         SELECT process_date FROM stream3
     )
     SELECT * FROM common_dates;
     

  2. 针对这些共同日期，从三个Stream中读取对应数据，调用存储过程执行MERGE INTO操作。
  3. 提交事务。
• 原理：事务内仅读取了符合条件的日期数据，未被读取的（仅2个Stream存在的日期）不会被标记为消费，会保留在Stream中，等待下次Task触发时再检查处理。

2. 使用AT时间点查询+显式消费（进阶原生特性） #

如果需要更精细的控制，可以先固定Stream的快照时间，再处理数据，最后显式消费已处理的部分：

• 记录当前快照时间：

  SET snapshot_ts = CURRENT_TIMESTAMP();
  

• 查询三个Stream在该快照时间点的共同日期：

  WITH stream1_dates AS (SELECT process_date FROM stream1 AT($snapshot_ts)),
       stream2_dates AS (SELECT process_date FROM stream2 AT($snapshot_ts)),
       stream3_dates AS (SELECT process_date FROM stream3 AT($snapshot_ts)),
       common_dates AS (SELECT * FROM stream1_dates INTERSECT stream2_dates INTERSECT stream3_dates)
  SELECT * FROM common_dates;
  

• 针对这些共同日期，读取三个Stream在该快照时间点的数据，调用存储过程处理。
• 处理完成后，显式消费Stream中已处理的变更：

  ALTER STREAM stream1 CONSUME;
  ALTER STREAM stream2 CONSUME;
  ALTER STREAM stream3 CONSUME;
  

• 注意：这种方式需要确保处理的是快照时间点的全部符合条件的数据，否则未处理的部分会被一起消费，所以推荐配合事务使用。

3. 临时表暂存Stream数据（备选方案） #

如果上述原生特性不好适配现有逻辑，可以先将Stream数据暂存到临时表，再基于临时表处理：

• 先将三个Stream的所有变更数据导入临时表：

  CREATE OR REPLACE TEMP TABLE temp_s1 AS SELECT * FROM stream1;
  CREATE OR REPLACE TEMP TABLE temp_s2 AS SELECT * FROM stream2;
  CREATE OR REPLACE TEMP TABLE temp_s3 AS SELECT * FROM stream3;
  

• 此时Stream的所有数据已被消费，但临时表保留了完整数据，你可以自由筛选共同日期调用存储过程处理；未处理的日期数据如果需要后续处理，可以将临时表改为持久化的暂存表，并添加process_date和处理状态标记。

• 避免在事务外单独查询Stream的全量数据，否则会误消费未处理的变更记录。
• Stream仅能捕获源表的增量变更，无法手动插入或恢复已消费的数据，所以处理逻辑必须确保只有已完成处理的记录才被标记为消费。

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