如何通过Snowpark异步执行表中存储的DDL语句？

方案一：Snowpark Python结合异步IO并发执行 #

利用Python的asyncio和Snowflake异步连接器，同时提交所有DDL语句到Snowflake后台并发执行，避免客户端逐行等待。

步骤与代码 #

1. 建立Snowpark会话并获取DDL列表

import snowflake.snowpark as snowpark
from snowflake.snowpark.session import Session
import asyncio
from snowflake.connector import connect

# 初始化Snowpark会话
def init_session():
    conn_params = {
        "account": "你的账户名",
        "user": "你的用户名",
        "password": "你的密码",
        "role": "你的角色",
        "warehouse": "你的仓库",
        "database": "目标数据库",
        "schema": "目标 schema"
    }
    return Session.builder.configs(conn_params).create()

session = init_session()

# 从表中提取所有DDL语句
ddl_records = session.sql("SELECT DDL_TEXT FROM DDL_STATEMENTS").collect()
ddl_statements = [row.DDL_TEXT for row in ddl_records]

2. 异步执行DDL的核心逻辑

async def run_single_ddl(ddl_text, conn_params):
    # 使用Snowflake异步连接器执行单条DDL
    async with connect(**conn_params) as conn:
        async with conn.cursor() as cur:
            await cur.execute(ddl_text)
            return f"执行成功: {ddl_text[:50]}..."

async def batch_execute_ddl():
    conn_params = {
        "account": "你的账户名",
        "user": "你的用户名",
        "password": "你的密码",
        "role": "你的角色",
        "warehouse": "你的仓库",
        "database": "目标数据库",
        "schema": "目标 schema"
    }
    # 创建所有异步任务
    tasks = [run_single_ddl(ddl, conn_params) for ddl in ddl_statements]
    # 并发执行所有任务
    execution_results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # 输出执行结果
    for idx, result in enumerate(execution_results):
        if isinstance(result, Exception):
            print(f"DDL {idx+1} 执行失败: {str(result)}")
        else:
            print(result)

# 启动异步执行
asyncio.run(batch_execute_ddl())

方案二：Snowpark存储过程结合异步任务 #

通过Snowflake任务（Task）实现后台异步执行DDL，无需客户端保持连接。存储过程会为每条DDL创建一个临时任务，任务执行DDL后自动删除自身。

存储过程代码 #

CREATE OR REPLACE PROCEDURE EXECUTE_DDL_ASYNC()
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'execute_ddl_batch'
AS $$
import snowflake.snowpark as snowpark

def execute_ddl_batch(session: snowpark.Session):
    # 获取DDL语句列表
    ddl_rows = session.sql("SELECT DDL_TEXT FROM DDL_STATEMENTS").collect()
    submitted_tasks = []
    
    for idx, row in enumerate(ddl_rows):
        ddl_content = row.DDL_TEXT
        # 生成唯一临时任务名
        task_name = f"TEMP_DDL_TASK_{idx}_{session.get_current_timestamp().strftime('%Y%m%d%H%M%S')}"
        # 构建创建任务的SQL
        task_sql = f"""
        CREATE OR REPLACE TASK {task_name}
        WAREHOUSE = YOUR_WAREHOUSE_NAME
        SCHEDULE = 'IMMEDIATE'
        AS
        BEGIN
            {ddl_content};
            DROP TASK IF EXISTS {task_name};
        END;
        """
        # 创建并启动任务
        session.sql(task_sql).collect()
        session.sql(f"ALTER TASK {task_name} RESUME").collect()
        submitted_tasks.append(task_name)
    
    return f"已提交 {len(submitted_tasks)} 个异步DDL任务: {', '.join(submitted_tasks)}"
$$;

调用存储过程 #

CALL EXECUTE_DDL_ASYNC();

关键注意事项 #

• 权限要求：执行方案二需要CREATE TASK、ALTER TASK权限，以及仓库的使用权限；方案一需要足够的连接并发配额。
• 依赖处理：如果DDL存在依赖关系（如先创建file_format再创建视图），需提前排序DDL语句，避免异步执行时出现依赖错误。
• 并发限制：仓库的并发查询数会影响异步执行的效率，需确保仓库有足够的并发能力。
• 错误处理：方案一中的return_exceptions=True会捕获异常并继续执行其他DDL，可根据需求调整；方案二可通过查询TASK_HISTORY查看任务执行状态。

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