如何基于S3多文件夹批量在Snowflake创建并加载Parquet表？

针对你有10000+个表目录的场景，可以通过以下步骤实现批量自动化处理：

1. 提取所有表目录名称 #

首先从stage中递归列出所有文件，通过正则提取出每个表的目录名（即table1、table2这类层级）：

SELECT DISTINCT 
  REGEXP_EXTRACT(RELATIVE_PATH, '^([^/]+)/', 1) AS TABLE_NAME
FROM TABLE(LIST(@estage, RECURSIVE => TRUE))
WHERE RELATIVE_PATH LIKE '%/part-%.parquet';

这个查询会过滤出所有包含Parquet数据文件的表目录，确保只处理有效的表路径。

2. 编写批量处理存储过程 #

基于你现有的单表处理逻辑，编写一个存储过程遍历所有表目录，自动执行表创建和数据加载：

CREATE OR REPLACE PROCEDURE BATCH_CREATE_AND_LOAD_TABLES()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    table_cursor CURSOR FOR
        SELECT DISTINCT REGEXP_EXTRACT(RELATIVE_PATH, '^([^/]+)/', 1) AS TABLE_NAME
        FROM TABLE(LIST(@estage, RECURSIVE => TRUE))
        WHERE RELATIVE_PATH LIKE '%/part-%.parquet';
    v_table_name VARCHAR;
    v_create_sql VARCHAR;
    v_copy_sql VARCHAR;
BEGIN
    FOR v_table_name IN table_cursor DO
        -- 生成创建表的动态SQL
        v_create_sql := 'CREATE OR REPLACE TABLE ' || v_table_name || '
USING TEMPLATE
(
    SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
    FROM TABLE(
        INFER_SCHEMA(
            LOCATION => ''' || '@estage/' || v_table_name || '''
            , FILE_FORMAT => ''parquet_format''
            , IGNORE_CASE => TRUE
        )
    )
);';
        
        -- 执行表创建
        EXECUTE IMMEDIATE v_create_sql;
        
        -- 生成数据加载的动态SQL
        v_copy_sql := 'COPY INTO ' || v_table_name || '
FROM @estage/' || v_table_name || '
FILE_FORMAT = (FORMAT_NAME = ''parquet_format'')
RECURSIVE = TRUE;';
        
        -- 执行数据加载
        EXECUTE IMMEDIATE v_copy_sql;
    END FOR;
    
    RETURN '所有表创建及加载完成';
EXCEPTION
    WHEN OTHERS THEN
        RETURN '处理失败：' || SQLERRM || '（失败表：' || v_table_name || '）';
END;
$$;

3. 执行批量处理 #

调用存储过程即可启动全量表的创建和加载：

CALL BATCH_CREATE_AND_LOAD_TABLES();

关键注意事项 #

• 权限配置：确保执行存储过程的角色拥有CREATE TABLE、USAGE（针对stage和file format）、SELECT（针对LIST操作）的权限
• 错误排查：建议新增日志表记录每个表的处理状态，方便定位失败案例，比如：

  CREATE TABLE TABLE_LOAD_LOG (
      TABLE_NAME VARCHAR,
      STATUS VARCHAR,
      ERROR_MESSAGE VARCHAR,
      LOAD_TIMESTAMP TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
  );
  

  然后在存储过程的循环和异常块中插入日志记录
• 性能优化：针对10000+表的规模，可考虑将存储过程拆分为分批处理，或者调整Snowflake的仓库大小提升处理速度
• 重复执行防护：如果需要保留已有表的数据，可修改创建表的逻辑为IF NOT EXISTS，避免覆盖现有数据

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