向Snowflake stage推送数据时如何自动触发Snowflake task

完全可以实现，目前有两种成熟的落地方式，可根据你的Snowflake版本选择：

方案1：Stage事件触发器（推荐，支持Standard Edition及以上版本） #

这是最轻量化的方案，无需额外中间组件，直接监听Stage的新文件写入事件触发Task，步骤如下：

• 首先配置所需权限（使用ACCOUNTADMIN角色执行）：

-- 替换尖括号内的内容为你实际的资源名称
GRANT EXECUTE TASK ON ACCOUNT TO ROLE <你的操作角色>;
GRANT CREATE EVENT TRIGGER ON SCHEMA <目标Schema名> TO ROLE <你的操作角色>;
GRANT READ ON STAGE <你的Internal Stage名> TO ROLE <你的操作角色>;

• 封装你之前手动执行的查询逻辑为Task：

CREATE OR REPLACE TASK AUTO_PROCESS_JSON_TASK
WAREHOUSE = <你的计算仓库名>
-- 此处调度规则仅为满足Task创建要求，后续由事件触发时不生效
SCHEDULE = '1 minute'
AS
-- 此处替换为你实际的业务查询逻辑，例如JSON解析、入库等
SELECT * FROM @<你的Internal Stage名> (FILE_FORMAT => '你的JSON格式定义名');

• 创建事件触发器监听Stage新文件上传事件，触发上述Task：

CREATE OR REPLACE EVENT TRIGGER STAGE_UPLOAD_TRIGGER
ON STAGE <你的Internal Stage名>
WHEN EVENT_TYPE = 'OBJECT_CREATED'
AS
EXECUTE TASK AUTO_PROCESS_JSON_TASK;

注意：该方案仅触发新文件写入事件，文件覆盖、删除操作不会触发任务

方案2：Snowpipe + Stream + Task 全版本兼容方案 #

如果你使用的Snowflake版本不支持事件触发器，可使用该兼容方案，原理是通过Snowpipe自动摄取Stage新文件，再通过Stream捕获数据增量，最后由Task监听增量自动执行业务逻辑：

• 首先创建JSON文件格式、存储原始数据的中间表：

CREATE OR REPLACE FILE FORMAT JSON_FORMAT
TYPE = JSON
STRIP_OUTER_ARRAY = TRUE;

CREATE OR REPLACE TABLE JSON_RAW_TEMP (
    RAW_DATA VARIANT,
    LOAD_TIME TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
);

• 创建自动摄取的Snowpipe，监听Stage新文件写入并自动写入中间表：

CREATE OR REPLACE PIPE JSON_AUTO_INGEST_PIPE
AUTO_INGEST = TRUE
AS
COPY INTO JSON_RAW_TEMP(RAW_DATA)
FROM @<你的Internal Stage名>
FILE_FORMAT = JSON_FORMAT;

• 为中间表创建仅追加模式的Stream，捕获新写入的增量数据：

CREATE OR REPLACE STREAM JSON_RAW_STREAM ON TABLE JSON_RAW_TEMP APPEND_ONLY = TRUE;

• 创建自动执行的Task，仅当Stream有增量数据时执行业务逻辑：

CREATE OR REPLACE TASK AUTO_PROCESS_JSON_TASK
WAREHOUSE = <你的计算仓库名>
SCHEDULE = '1 minute'
WHEN SYSTEM$STREAM_HAS_DATA('JSON_RAW_STREAM')
AS
-- 此处替换为你实际的业务查询逻辑
INSERT INTO 你的业务表(字段1, 字段2)
SELECT RAW_DATA:字段1::STRING, RAW_DATA:字段2::NUMBER FROM JSON_RAW_STREAM;

注意：使用该方案需要按照你Snowflake账号对应的云厂商后端，配置Internal Stage的事件通知集成到Snowpipe，即可开启自动摄取能力

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