如何让Snowpipe仅加载S3阶段存储中的新数据至指定表

要让新建的Snowpipe只处理@GAME.GAME_SAVES_STAGE/game=xyz路径下的新增数据（跳过已通过克隆获取的历史数据），最直接的方案是利用Snowpipe的START_TIME参数，在创建管道时明确指定起始处理时间。

具体操作步骤 #

1. 确认历史文件的最晚修改时间
   先查询目标阶段下所有历史文件的最后修改时间，确保后续指定的起始时间晚于这个时间点：

   LIST @GAME.GAME_SAVES_STAGE/game=xyz;
   

   从返回结果中找到LAST_MODIFIED列的最大值，或者直接用当前时间（前提是所有历史文件已经完成上传）。

2. 带START_TIME参数创建管道
   创建管道时，开启AUTO_INGEST的同时，指定START_TIME为你确认的时间点，这样管道只会处理该时间之后新增的文件：

   CREATE OR REPLACE PIPE GAME.XYZ_GAME_SAVES_PIPE
   AUTO_INGEST = TRUE
   START_TIME => CURRENT_TIMESTAMP() -- 或替换为查到的最晚历史文件修改时间
   AS
   COPY INTO GAME.XYZ_GAME_SAVES_RAW
   FROM @GAME.GAME_SAVES_STAGE/game=xyz
   FILE_FORMAT = (FORMAT_NAME = YOUR_FILE_FORMAT) -- 替换为你的实际文件格式
   PATTERN = '.*\.json' -- 根据你的文件类型调整匹配规则
   ;
   

原理说明 #

START_TIME参数会告诉Snowpipe，只处理该时间点之后被S3事件通知触发的文件，而早于该时间的历史文件不会被纳入管道的处理队列，完美适配你已经通过克隆获取历史数据的场景。

备选方案（如果已误创建管道） #

如果已经创建了未指定START_TIME的管道，且它开始加载历史数据，可以按以下步骤修正：

1. 暂停管道：

   ALTER PIPE GAME.XYZ_GAME_SAVES_PIPE SET PIPE_EXECUTION_PAUSED = TRUE;
   

2. 删除现有管道：

   DROP PIPE GAME.XYZ_GAME_SAVES_PIPE;
   

3. 重新按上述方法带START_TIME参数创建新管道。

内容的提问来源于stack exchange，提问作者Anum Sheraz