Snowflake仓库多调度场景下如何实现同一时间仅运行一个任务

以下是几种确保Snowflake中多个调度任务同一时间仅运行一个的可行方案：

1. 基于锁表的自定义互斥机制 #

创建专门的锁表，每个任务执行前先尝试原子性获取锁，成功则继续执行，否则直接退出。这种方案灵活性高，适用于任意任务组合的互斥场景。

实现步骤： #

• 创建锁表并初始化锁记录：

CREATE TABLE IF NOT EXISTS TASK_LOCK (
    LOCK_ID VARCHAR(50) DEFAULT 'MAIN_LOCK',
    IS_LOCKED BOOLEAN DEFAULT FALSE,
    LOCK_ACQUIRED_TIME TIMESTAMP,
    PRIMARY KEY (LOCK_ID)
);
INSERT INTO TASK_LOCK (LOCK_ID) VALUES ('MAIN_LOCK') ON DUPLICATE KEY UPDATE LOCK_ID = LOCK_ID;

• 在任务执行的存储过程中添加锁校验逻辑：

CREATE OR REPLACE PROCEDURE RUN_TASK_LOGIC()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    v_lock_acquired BOOLEAN DEFAULT FALSE;
BEGIN
    -- 原子性获取锁
    UPDATE TASK_LOCK
    SET IS_LOCKED = TRUE, LOCK_ACQUIRED_TIME = CURRENT_TIMESTAMP()
    WHERE LOCK_ID = 'MAIN_LOCK' AND IS_LOCKED = FALSE;

    -- 检查锁获取结果
    SELECT IS_LOCKED INTO v_lock_acquired
    FROM TASK_LOCK WHERE LOCK_ID = 'MAIN_LOCK';

    IF NOT v_lock_acquired THEN
        RETURN 'Lock held by another task, exiting.';
    END IF;

    -- 执行任务核心逻辑
    -- ...

    -- 释放锁
    UPDATE TASK_LOCK
    SET IS_LOCKED = FALSE
    WHERE LOCK_ID = 'MAIN_LOCK';

    RETURN 'Task completed successfully.';
EXCEPTION
    WHEN OTHERS THEN
        -- 异常时强制释放锁
        UPDATE TASK_LOCK
        SET IS_LOCKED = FALSE
        WHERE LOCK_ID = 'MAIN_LOCK';
        RETURN 'Task failed: ' || SQLERRM;
END;
$$;

• 将任务指向该存储过程，确保执行前完成锁校验。

2. 依赖链+单触发主任务 #

将所有需要互斥的子任务设置为无独立调度，通过一个主调度任务控制同一时间仅触发一个子任务执行。适合任务需按顺序或条件触发的场景。

实现步骤： #

• 创建无调度的子任务：

CREATE OR REPLACE TASK TASK1
WAREHOUSE = YOUR_WH
AS CALL TASK1_LOGIC();

CREATE OR REPLACE TASK TASK2
WAREHOUSE = YOUR_WH
AS CALL TASK2_LOGIC();

• 创建主调度任务与控制存储过程：

CREATE OR REPLACE PROCEDURE SCHEDULE_MUTEX_TASKS()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    v_running_task VARCHAR;
BEGIN
    -- 检查是否有子任务正在运行
    SELECT NAME INTO v_running_task
    FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
        SCHEDULED_TIME_RANGE_START => DATEADD('minute', -5, CURRENT_TIMESTAMP()),
        STATE => 'RUNNING'
    ))
    WHERE NAME IN ('TASK1', 'TASK2')
    LIMIT 1;

    IF v_running_task IS NOT NULL THEN
        RETURN 'Task ' || v_running_task || ' is running, skipping scheduling.';
    END IF;

    -- 按轮询/优先级逻辑选择子任务触发（示例为触发TASK1）
    CALL SYSTEM$TASK_RESUME('TASK1');
    RETURN 'Triggered TASK1 successfully.';
END;
$$;

CREATE OR REPLACE TASK MAIN_SCHEDULER_TASK
WAREHOUSE = YOUR_WH
SCHEDULE = 'USING CRON 0 * * * * UTC' -- 自定义调度频率
AS CALL SCHEDULE_MUTEX_TASKS();

3. 限制仓库并发数 #

将所有互斥任务指定到专属仓库，设置仓库最大并发级别为1，强制所有任务串行执行。方案简单直接，但需确保仓库仅用于互斥任务。

实现步骤： #

• 修改仓库并发限制：

ALTER WAREHOUSE YOUR_MUTEX_WH
SET MAX_CONCURRENCY_LEVEL = 1;

• 将任务关联到该仓库：

ALTER TASK TASK1 SET WAREHOUSE = YOUR_MUTEX_WH;
ALTER TASK TASK2 SET WAREHOUSE = YOUR_MUTEX_WH;

注意：该仓库的所有任务都会被限制为串行执行，需避免非互斥任务使用此仓库。

4. 任务执行前检查运行状态 #

在每个任务的逻辑中，先查询Snowflake任务历史视图，检查是否有其他互斥任务正在运行，若有则直接退出。

示例代码（存储过程中添加检查）： #

CREATE OR REPLACE PROCEDURE TASK1_LOGIC()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    v_running_count INT;
BEGIN
    -- 检查指定任务列表的运行状态
    SELECT COUNT(*) INTO v_running_count
    FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
        SCHEDULED_TIME_RANGE_START => DATEADD('minute', -5, CURRENT_TIMESTAMP()),
        STATE => 'RUNNING'
    ))
    WHERE NAME IN ('TASK1', 'TASK2', 'TASK3');

    IF v_running_count > 0 THEN
        RETURN 'Other mutex tasks are running, exiting.';
    END IF;

    -- 执行任务核心逻辑
    -- ...

    RETURN 'Task1 completed.';
END;
$$;

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