如何在Snowflake中创建存储过程运行含CTE和JS UDF的SQL脚本并定时更新结果集

1. 提前创建依赖资源 #

1.1 创建时间差计算JS UDF #

该UDF只需创建一次，后续无修改无需重复部署：

create or replace function tsrange_intersection(s string, e string)
RETURNS double
LANGUAGE JAVASCRIPT
AS
$$
let minutes = 0
start = new Date(S)
end = new Date(E)
let t = start
while(t < end) {
    if ([1, 2, 3, 4, 5].includes(t.getDay())
      && [9, 10, 11, 12, 13, 14, 15, 16].includes(t.getHours())) {
        minutes += 1
    }
    t = new Date(t.getTime() + 60*1000);
}
return minutes
$$;

1.2 确认结果表（首次执行创建即可） #

可以选择直接在存储过程中自动创建/覆盖表，也可以提前创建固定结构的表：

-- 提前创建结果表（适合做增量更新的场景）
CREATE OR REPLACE TABLE status_time_diff (
    id STRING,
    status STRING,
    start_timestamp TIMESTAMP,
    end_timestamp TIMESTAMP,
    time_diff DOUBLE
);

2. 创建存储过程 #

将你的整个查询逻辑合并为单条SQL（移除RESULT_SCAN依赖，避免跨查询出错），存储过程支持全量覆盖更新或增量更新，以下是全量覆盖的示例（适合数据量不大的场景，每次执行直接生成最新完整结果）：

CREATE OR REPLACE PROCEDURE sp_calculate_status_time_diff()
RETURNS STRING
LANGUAGE SQL
AS
$$
BEGIN
  -- 直接覆盖更新结果表，也可以根据需求改为TRUNCATE + INSERT或者MERGE增量更新
  CREATE OR REPLACE TABLE status_time_diff AS
  WITH dataset AS (
    -- 注意：此处替换为你实际的业务源表，示例是测试用的values构造数据
    select $1 id, $2 status, $3 created_at, $4 completed_at
    from values
      ('A','created','2021-07-15 10:30:00'::timestamp, NULL), ('A','missing_info','2021-07-15 11:10:00'::timestamp,NULL),
      ('A','pending','2021-07-15 12:05:00'::timestamp, NULL), ('A','successful','2021-07-15 16:05:00'::timestamp,'2021-07-15 17:05:00'::timestamp),
      ('B','created','2021-07-16 11:30:00'::timestamp, NULL), ('B',NULL,'2021-07-16 11:30:00'::timestamp, NULL),
      ('B','successful','2021-07-16 12:30:00'::timestamp, '2021-07-16 16:30:00'::timestamp) 
  ),
  cte1 AS (
    SELECT *
    FROM dataset
    WHERE id IS NOT NULL AND status IS NOT NULL 
    ORDER BY created_at 
  ),
  cte2 AS (
    SELECT *
    FROM cte1
    QUALIFY FIRST_VALUE(status) OVER (PARTITION BY id ORDER BY created_at) = 'created' 
  ),
  -- 原来的MATCH_RECOGNIZE逻辑直接作为子查询，不需要RESULT_SCAN
  cte3 AS (
    SELECT *
    FROM cte2
    MATCH_RECOGNIZE (
        PARTITION BY id
        ORDER BY created_at
        MEASURES MATCH_NUMBER() AS mn,
                 MATCH_SEQUENCE_NUMBER AS msn
        ALL ROWS PER MATCH
        PATTERN (c+i+|p+i+|ps+)
        DEFINE
            c AS status='created',
            i AS status='missing_info',
            p AS status='pending',
            s AS status='successful'
        ) mr
        QUALIFY (ROW_NUMBER() OVER (PARTITION BY mn, id ORDER BY msn) = 1)
              OR(ROW_NUMBER() OVER (PARTITION BY mn, id ORDER BY msn DESC) =1)
  ),
  cte4 AS (
    SELECT
        *,
        CASE WHEN status = 'successful' THEN IFNULL(completed_at, created_at) ELSE created_at END AS start_timestamp
    FROM cte3 
  ),
  cte5 AS (
    SELECT
      *,
      CASE WHEN status = 'successful' THEN COMPLETED_AT 
      ELSE IFNULL(LAG(start_timestamp, -1) OVER (PARTITION BY id ORDER BY start_timestamp), completed_at) END AS end_timestamp
    FROM cte4 
  )
  SELECT 
      id, status, start_timestamp, end_timestamp,
      tsrange_intersection(start_timestamp::string, end_timestamp::string) AS time_diff
  FROM cte5
  ORDER BY start_timestamp;

  RETURN '计算完成，结果已更新到status_time_diff表';
END;
$$;

创建完成后可以手动执行测试：CALL sp_calculate_status_time_diff();

3. 创建定时Task执行存储过程 #

以下是每天凌晨1点执行的示例，可根据需求修改调度频率：

CREATE OR REPLACE TASK task_calculate_status_time_diff
  WAREHOUSE = 你的计算仓库名 -- 替换为你实际用的仓库
  SCHEDULE = 'USING CRON 0 1 * * * UTC' -- 可修改Cron表达式调整调度时间，也可直接写如'1 DAY'这类简单频率
  COMMENT = '每日定时计算状态时间差'
AS
  CALL sp_calculate_status_time_diff();

Task创建后默认是暂停状态，需要手动开启：

ALTER TASK task_calculate_status_time_diff RESUME;

注意事项 #

• 执行Task的账号需要有EXECUTE TASK权限，以及对应仓库、存储过程、表的读写权限
• 如果需要做增量更新，只需修改存储过程中的逻辑，增加源表的时间过滤条件，用MERGE语句更新结果表即可
• Cron表达式默认是UTC时区，可自行修改为业务所需时区，比如SCHEDULE = 'USING CRON 0 1 * * * Asia/Shanghai'

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