如何实现Snowflake中从sp1并行调用存储过程sp2？

问题背景 #

你用JavaScript编写了存储过程sp1，通过循环顺序调用sp2处理表列表中的数据，现在需要改为并行调用sp2提升效率，尝试过Snowpark Python但未实现并发执行。

可行解决方案 #

方案1：利用Snowflake任务（Task）实现并行触发 #

Snowflake任务支持独立调度执行，可通过在sp1中动态触发多个任务来并行调用sp2。

操作步骤：

1. 在sp1中为每个待处理的表动态生成临时任务，任务逻辑为调用sp2并传入对应表名。
2. 批量触发这些任务，任务会在后台并行执行，执行后可清理临时任务避免残留。

代码示例：

create procedure sp1()
language javascript
$$
  const tbllist = ['tb1','tbl2','tbl3'];
  const taskPrefix = 'TASK_SP2_';
  const warehouseName = 'YOUR_WAREHOUSE_NAME';
  
  for (let tbl of tbllist) {
    const taskName = taskPrefix + tbl;
    // 创建一次性执行的临时任务
    const createTaskSql = `
      CREATE OR REPLACE TASK ${taskName}
      WAREHOUSE = ${warehouseName}
      SCHEDULE = 'USING CRON * * * * * UTC'
      AS CALL sp2('${tbl}');
    `;
    snowflake.execute({sqlText: createTaskSql});
    
    // 触发任务立即执行
    snowflake.execute({sqlText: `EXECUTE TASK ${taskName};`});
    
    // 执行后删除临时任务
    snowflake.execute({sqlText: `DROP TASK IF EXISTS ${taskName};`});
  }
$$;

注意：执行sp1的角色需要拥有创建、执行、删除任务的权限，且指定仓库的并发资源需足够支撑并行任务。

方案2：使用Snowpark Python异步API实现并发调用 #

Snowpark Python提供异步执行方法，可在存储过程中提交多个异步调用，突破单线程限制实现sp2并行执行。

代码示例：

from snowflake.snowpark import Session
import asyncio

def sp1(session: Session):
    tbllist = ['tb1', 'tbl2', 'tbl3']
    async_tasks = []
    
    # 批量提交异步调用任务
    for tbl in tbllist:
        async_task = session.sql(f"CALL sp2('{tbl}')").collect_async()
        async_tasks.append(async_task)
    
    # 等待所有异步任务完成
    asyncio.run(asyncio.gather(*async_tasks))
    return "所有并行任务执行完成"

# 注册存储过程
session.sproc.register(
    func=sp1,
    name="sp1",
    is_permanent=True,
    stage_location="@your_stage",
    packages=['snowflake-snowpark-python']
)

注意：需使用v1.10及以上版本的Snowpark Python，同时仓库的并发数配置要匹配并行任务数量。

方案3：外部程序多会话并行调用（适合外部驱动场景） #

如果逻辑允许从外部应用触发，可通过Python/Java等外部程序创建多个Snowflake会话，直接并行调用sp2，这种方式不受存储过程单线程限制，并行度更高。

简化示例（Python外部脚本）：

import snowflake.connector
from concurrent.futures import ThreadPoolExecutor

def call_sp2(tbl):
    conn = snowflake.connector.connect(
        user='YOUR_USER',
        password='YOUR_PASSWORD',
        account='YOUR_ACCOUNT',
        warehouse='YOUR_WAREHOUSE'
    )
    cursor = conn.cursor()
    cursor.execute(f"CALL sp2('{tbl}')")
    cursor.close()
    conn.close()

tbllist = ['tb1','tbl2','tbl3']
# 线程池并行执行
with ThreadPoolExecutor(max_workers=3) as executor:
    executor.map(call_sp2, tbllist)

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