如何通过Snowflake从S3多文件批量加载至多张表（免重复写COPY命令）

1. 用Snowflake脚本批量生成并执行COPY命令 #

通过定义表与S3文件的映射关系，编写SQL脚本自动循环执行对应表的COPY命令，无需手动逐个编写。

首先创建临时表存储映射关系：

CREATE OR REPLACE TEMPORARY TABLE TABLE_FILE_MAPPING (
    TABLE_NAME VARCHAR,
    S3_FILE_PATH VARCHAR,
    FILE_FORMAT_NAME VARCHAR
);

-- 插入所有需要加载的表与文件对应规则
INSERT INTO TABLE_FILE_MAPPING VALUES
('TABLE_A', 's3://your-bucket/path/to/file_a.csv', 'MY_CSV_FORMAT'),
('TABLE_B', 's3://your-bucket/path/to/file_b.parquet', 'MY_PARQUET_FORMAT'),
('TABLE_C', 's3://your-bucket/path/to/file_c.json', 'MY_JSON_FORMAT');

然后用脚本循环执行COPY：

DECLARE
    CUR CURSOR FOR SELECT TABLE_NAME, S3_FILE_PATH, FILE_FORMAT_NAME FROM TABLE_FILE_MAPPING;
    v_table VARCHAR;
    v_path VARCHAR;
    v_format VARCHAR;
BEGIN
    FOR rec IN CUR DO
        v_table := rec.TABLE_NAME;
        v_path := rec.S3_FILE_PATH;
        v_format := rec.FILE_FORMAT_NAME;
        
        EXECUTE IMMEDIATE 'COPY INTO ' || v_table || ' FROM ''' || v_path || ''' FILE_FORMAT = ' || v_format;
    END FOR;
END;

2. 封装存储过程+任务实现定时批量加载 #

如果需要定期执行批量加载，可将上述逻辑封装为存储过程，再通过Snowflake任务调度执行。

创建存储过程：

CREATE OR REPLACE PROCEDURE BULK_LOAD_TABLES()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    CUR CURSOR FOR SELECT TABLE_NAME, S3_FILE_PATH, FILE_FORMAT_NAME FROM TABLE_FILE_MAPPING;
    v_table VARCHAR;
    v_path VARCHAR;
    v_format VARCHAR;
BEGIN
    FOR rec IN CUR DO
        v_table := rec.TABLE_NAME;
        v_path := rec.S3_FILE_PATH;
        v_format := rec.FILE_FORMAT_NAME;
        
        EXECUTE IMMEDIATE 'COPY INTO ' || v_table || ' FROM ''' || v_path || ''' FILE_FORMAT = ' || v_format;
    END FOR;
    RETURN '批量加载完成';
END;
$$;

创建并启动任务（示例为每天UTC 0点执行）：

CREATE OR REPLACE TASK BULK_LOAD_TASK
WAREHOUSE = YOUR_WAREHOUSE_NAME
SCHEDULE = 'USING CRON 0 0 * * * UTC'
AS
CALL BULK_LOAD_TABLES();

ALTER TASK BULK_LOAD_TASK RESUME;

3. 外部表+Merge（适合增量加载） #

如果是增量更新的文件，可先创建对应S3文件的外部表，再批量通过Merge同步到目标表：

DECLARE
    CUR CURSOR FOR SELECT TABLE_NAME, S3_FILE_PATH, FILE_FORMAT_NAME FROM TABLE_FILE_MAPPING;
    v_table VARCHAR;
    v_path VARCHAR;
    v_format VARCHAR;
    v_external_table VARCHAR;
BEGIN
    FOR rec IN CUR DO
        v_table := rec.TABLE_NAME;
        v_path := rec.S3_FILE_PATH;
        v_format := rec.FILE_FORMAT_NAME;
        v_external_table := 'EXT_' || v_table;
        
        -- 创建外部表
        EXECUTE IMMEDIATE 'CREATE OR REPLACE EXTERNAL TABLE ' || v_external_table || 
                          ' LOCATION = ''' || v_path || ''' FILE_FORMAT = ' || v_format;
        
        -- Merge同步（假设目标表主键为ID）
        EXECUTE IMMEDIATE 'MERGE INTO ' || v_table || ' t ' ||
                          'USING ' || v_external_table || ' e ' ||
                          'ON t.ID = e.ID ' ||
                          'WHEN MATCHED THEN UPDATE SET * ' ||
                          'WHEN NOT MATCHED THEN INSERT *';
    END FOR;
END;

注意事项 #

• 确保Snowflake角色拥有S3访问、表操作、脚本/存储过程执行等权限
• 映射表中的文件路径和格式需准确匹配，避免执行报错
• 可在脚本中添加异常捕获逻辑，记录加载错误日志

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