如何自动化加载AWS S3中的大尺寸.sql文件至Snowflake

一、核心方案：Snowflake外部Stage+存储过程拆分执行 #

针对EXECUTE IMMEDIATE FROM的10MB限制，通过将S3桶挂载为Snowflake外部Stage，再用存储过程读取文件内容、按语句拆分后逐段执行，绕开大小限制。

步骤1：创建关联S3的外部Stage #

CREATE OR REPLACE STAGE s3_sql_stage
URL = 's3://your-bucket/path/to/sql/files/'
CREDENTIALS = (AWS_KEY_ID = 'your-aws-key' AWS_SECRET_KEY = 'your-aws-secret');

如果用IAM角色认证，替换CREDENTIALS为IAM_ROLE = 'arn:aws:iam::xxx:role/xxx'

步骤2：编写拆分执行的存储过程 #

CREATE OR REPLACE PROCEDURE RUN_LARGE_SQL_FROM_S3(file_name VARCHAR)
RETURNS VARCHAR
LANGUAGE JAVASCRIPT
EXECUTE AS CALLER
AS
$$
    // 读取Stage中的SQL文件
    var sql_file = snowflake.execute({sqlText: `SELECT $1 FROM @s3_sql_stage/` + FILE_NAME});
    sql_file.next();
    var full_sql = sql_file.getColumnValue(1);
    
    // 按分号拆分SQL（规避字符串内的分号）
    var sql_statements = full_sql.split(/;(?=(?:[^'"]|'[^']*'|"[^"]*")*$)/);
    
    // 逐段执行语句
    for (var stmt of sql_statements) {
        stmt = stmt.trim();
        if (stmt !== '') {
            try {
                snowflake.execute({sqlText: stmt});
            } catch (err) {
                return `执行失败：语句片段 - ${stmt.slice(0, 100)}...，错误：${err.message}`;
            }
        }
    }
    return '所有SQL语句执行完成';
$$;

步骤3：调用存储过程执行 #

CALL RUN_LARGE_SQL_FROM_S3('your-large-sql-file.sql');

二、Python自动化执行方案 #

通过Snowflake Python连接器+AWS SDK读取S3文件，拆分后批量执行，适合外部调度场景。

代码示例 #

import snowflake.connector
import boto3
import re

# 读取S3中的SQL文件
s3 = boto3.client('s3')
response = s3.get_object(Bucket='your-bucket', Key='path/to/your-sql-file.sql')
full_sql = response['Body'].read().decode('utf-8')

# 拆分SQL语句（处理字符串内的分号）
sql_statements = re.split(r';(?=(?:[^"\']|["\'][^"\']*["\'])*$)', full_sql)

# 连接Snowflake并执行
conn = snowflake.connector.connect(
    user='your-snowflake-user',
    password='your-password',
    account='your-account-id',
    warehouse='your-warehouse',
    database='your-db',
    schema='your-schema'
)
cursor = conn.cursor()

try:
    for stmt in sql_statements:
        stmt = stmt.strip()
        if stmt:
            cursor.execute(stmt)
    print('所有SQL执行完成')
finally:
    cursor.close()
    conn.close()

可将代码部署为AWS Lambda、Airflow任务实现定时自动化

三、Snowflake Notebook自动化实现 #

完全可以通过Snowflake Notebook完成自动化：

• 在Notebook内直接挂载S3 Stage读取文件
• 用Python或SQL单元格编写拆分执行逻辑
• 配置Notebook的任务调度，定期触发执行
• 优势：无需外部环境，Snowflake生态内完成，便于监控和日志回溯

Notebook核心代码（Python单元格） #

import snowflake.snowpark as snowpark
import re

def main(session: snowpark.Session):
    # 读取Stage中的SQL文件
    df = session.read.text("@s3_sql_stage/your-large-sql-file.sql")
    full_sql = df.collect()[0][0]
    
    # 拆分SQL语句
    sql_statements = re.split(r';(?=(?:[^"\']|["\'][^"\']*["\'])*$)', full_sql)
    
    # 逐段执行
    for stmt in sql_statements:
        stmt = stmt.strip()
        if stmt:
            session.sql(stmt).collect()
    return "SQL执行完成"

四、关键注意事项 #

• 拆分SQL的正则表达式需规避字符串内的分号，防止误拆分
• 若存在单条超10MB的INSERT语句，建议改用COPY INTO（将数据文件存S3，批量导入效率远高于INSERT）
• 自动化执行时需添加错误捕获和日志记录，便于问题排查

内容的提问来源于stack exchange，提问作者s.bramblet