如何在Snowflake中实现单例模式避免存储过程并发执行？

我们需要实现单例模式，确保存储过程不会被同时多次运行。由于暂未找到Snowflake原生支持该功能的特性，因此考虑通过「Lock」表实现，批处理场景下秒级等待可接受。

现有实现方案 #

锁表结构 #

SHARED.LOCK(LOCK_NAME           STRING NOT NULL PRIMARY KEY
           ,SESSION_ID          STRING NOT NULL
           ,ACQUIRED_AT         TIMESTAMP_NTZ
           )

字段说明：

• LOCK_NAME强制转为大写，作为主键使用
• SESSION_ID为当前会话ID
• ACQUIRED_AT为锁获取时间，仅作信息参考

锁获取逻辑 #

创建存储过程来获取指定$LOCK_NAME的锁，当锁未被占有时，尝试更新锁记录写入当前会话信息，更新语句如下：

UPDATE SHARED.LOCK
  SET LOAD_ID = $LOAD_ID
     ,SESSION_ID = CURRENT_SESSION()
     ,ACQUIRED_AT = CURRENT_TIMESTAMP()
  WHERE LOCK_NAME = $LOCK_NAME
    AND SESSION_ID IS NULL;

为避免Snowflake乐观锁副作用，确保该存储过程不会在显式事务中被调用。
之后校验是否成功获取锁，校验语句如下：

SELECT 1
  FROM SHARED.LOCK
  WHERE LOCK_NAME = $LOCK_NAME
    AND LOAD_ID = $SESSION_ID;

如果查询返回记录，说明成功获取锁；如果未获取到，可以等待X秒后重试，最多重试指定次数。

锁释放逻辑 #

业务逻辑执行完成后，通过以下Update语句释放锁：

UPDATE SHARED.LOCK
  SET SESSION_ID = NULL
     ,ACQUIRED_AT = NULL
  WHERE LOCK_NAME = $LOCK_NAME
    AND SESSION_ID = $SESSION_ID;

待处理异常 #

还需要处理超时未释放、被已失效会话持有的锁等异常场景。

你提到的锁表方案本身是可行的，以下是更简单、容错性更强的实现方向：

1 简化锁逻辑，减少冗余操作 #

你现有代码里提到的LOAD_ID字段没有在表结构中定义，属于冗余设计，可以直接删除，用MERGE语句替代「更新+校验」的两步操作，一步完成锁获取判断，减少交互次数：

MERGE INTO SHARED.LOCK t
USING (SELECT UPPER($LOCK_NAME) AS LOCK_NAME) s
ON t.LOCK_NAME = s.LOCK_NAME
WHEN MATCHED AND t.SESSION_ID IS NULL THEN
  UPDATE SET SESSION_ID = CURRENT_SESSION(), ACQUIRED_AT = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
  INSERT (LOCK_NAME, SESSION_ID, ACQUIRED_AT)
  VALUES (s.LOCK_NAME, CURRENT_SESSION(), CURRENT_TIMESTAMP());

如果MERGE返回的影响行数为1，就代表成功获取锁，不需要额外执行SELECT校验，逻辑更简洁。

2 内置异常处理逻辑，无需额外运维 #

在获取锁的条件中直接加入超时、会话失效判断，不需要额外写定时任务清理无效锁：

-- 加在MERGE的MATCHED条件中
WHEN MATCHED AND (
  t.SESSION_ID IS NULL
  OR ACQUIRED_AT < DATEADD('minute', -30, CURRENT_TIMESTAMP()) -- 锁超过30分钟自动失效，时长可自定义
  OR NOT EXISTS (SELECT 1 FROM TABLE(INFORMATION_SCHEMA.SESSIONS()) WHERE SESSION_ID = t.SESSION_ID) -- 会话已失效自动释放锁
) THEN

3 最简单方案：用调度工具原生能力 #

如果你的存储过程是通过调度工具触发的，完全不需要自己实现锁逻辑：

• 如果用Snowflake原生Task调度，直接设置ALLOW_OVERLAPPING_EXECUTION = FALSE参数，即可禁止同一Task的实例重叠执行
• 如果用Airflow、dbt等第三方调度工具，设置任务最大并行数为1即可实现单例运行

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