Snowflake任务调度：如何结合CRON与非CRON实现数据就绪后触发存储过程

问题核心回顾 #

你遇到的报错是因为Snowflake任务不能同时设置SCHEDULE调度和AFTER前置任务，这是平台的固有限制。要实现「每小时整点启动，每分钟检查表中是否有新数据，有则执行存储过程并停止检查」的逻辑，可以通过Stream捕获数据变化+两级任务配合来解决。

具体实现步骤 #

1. 用Stream跟踪目标表的新增数据 #

先给需要监控的表创建一个仅捕获新增数据的Stream：

CREATE OR REPLACE STREAM target_table_stream
ON TABLE your_target_table -- 替换成你要监控的表名
APPEND_ONLY = TRUE; -- 只追踪插入操作，匹配你的需求

2. 创建小时级启动任务 #

这个任务仅负责在每小时整点激活检查任务，不处理业务逻辑：

CREATE OR REPLACE TASK tsk_hourly_launch
WAREHOUSE = your_warehouse -- 替换成你的仓库名
SCHEDULE = 'USING CRON 0 * * * * UTC' -- 每小时整点执行
AS
ALTER TASK tsk_minute_check RESUME; -- 激活每分钟检查的任务

3. 创建核心的每分钟检查任务 #

这个任务通过自身逻辑控制启停，无需设置前置任务：

CREATE OR REPLACE TASK tsk_minute_check
WAREHOUSE = your_warehouse
SCHEDULE = '1 minute' -- 每分钟执行一次
SUSPEND_TASK_AFTER_NUM_FAILURES = 0 -- 避免因无数据导致的"失败"自动暂停任务
AS
BEGIN
    -- 检查Stream里有没有未处理的新数据
    IF EXISTS (SELECT 1 FROM target_table_stream) THEN
        -- 调用你的存储过程
        CALL your_stored_procedure(); -- 替换成实际的存储过程名
        
        -- 执行完成后暂停检查任务，等下一小时再启动
        ALTER TASK tsk_minute_check SUSPEND;
        
        -- 可选：把Stream里的记录消费掉（比如存入历史表），避免重复检测
        INSERT INTO your_data_history_table SELECT * FROM target_table_stream;
        TRUNCATE TABLE target_table_stream;
    END IF;
END;

4. 初始化任务状态 #

首次创建后，需要先把检查任务暂停，让小时任务来激活它：

ALTER TASK tsk_minute_check SUSPEND;
ALTER TASK tsk_hourly_launch RESUME;

逻辑运行流程 #

1. 每小时整点，tsk_hourly_launch执行，将暂停状态的tsk_minute_check恢复，开始每分钟检查一次
2. 每次tsk_minute_check运行时：
   • 如果Stream里有新数据（说明目标表刚加载了数据）：调用存储过程，然后暂停自己，直到下一小时被激活
   • 如果没有新数据：直接结束，等待下一分钟再次检查
3. Stream会自动记录目标表的新增数据，直到你通过插入历史表或截断的方式消费掉这些记录

注意事项 #

• 确保Stream的APPEND_ONLY属性正确，如果你的数据加载包含更新/删除操作，需要调整Stream的类型
• 存储过程执行成功后一定要暂停检查任务，否则会每分钟重复执行
• 如果数据加载可能延迟超过一小时，可以调整小时任务的调度频率，或者在检查任务里增加超时逻辑
• 检查频率可根据实际需求调整（比如改成30秒一次，只需把SCHEDULE改成'30 seconds'即可）

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