如何并行执行SQL Server存储过程？ADF批量数据迁移优化需求

针对你提到的6亿级数据迁移场景，以下是SQL Server、ADF、Python、Spark四种并行执行子存储过程的实现方案：

一、SQL Server 端异步并行方案 #

方法1：利用xp_cmdshell异步调用子存储过程 #

通过xp_cmdshell启动独立的sqlcmd进程，每个进程执行一个子存储过程，实现并行。注意需要提前开启xp_cmdshell权限，并做好安全管控。

-- 开启xp_cmdshell（仅首次配置需要）
EXEC sp_configure 'show advanced options', 1;
RECONFIGURE;
EXEC sp_configure 'xp_cmdshell', 1;
RECONFIGURE;

-- 异步执行三个子存储过程（替换占位符为实际服务器/数据库/账号信息）
EXEC xp_cmdshell 'sqlcmd -S YourSqlServer -d YourDB -U YourUser -P YourPass -Q "EXEC proc1" -b', NO_OUTPUT;
EXEC xp_cmdshell 'sqlcmd -S YourSqlServer -d YourDB -U YourUser -P YourPass -Q "EXEC proc2" -b', NO_OUTPUT;
EXEC xp_cmdshell 'sqlcmd -S YourSqlServer -d YourDB -U YourUser -P YourPass -Q "EXEC proc3" -b', NO_OUTPUT;

方法2：SQL Agent 作业并行触发 #

创建三个独立的SQL Agent作业，每个作业绑定一个子存储过程，主存储过程中调用sp_start_job同时启动这三个作业：

-- 启动三个并行作业
EXEC msdb.dbo.sp_start_job N'Job_Exec_Proc1';
EXEC msdb.dbo.sp_start_job N'Job_Exec_Proc2';
EXEC msdb.dbo.sp_start_job N'Job_Exec_Proc3';

这种方式无需开启xp_cmdshell，权限管控更规范，适合生产环境。

二、Azure Data Factory (ADF) 管道并行方案 #

ADF原生支持活动并行，直接拆分原存储过程调用逻辑，用三个独立的「执行存储过程」活动并行执行：

1. 拆分活动：移除原调用主存储过程的活动，添加三个「执行存储过程」活动，分别配置调用proc1、proc2、proc3，按需传入表名等参数。
2. 设置并行：将三个活动的前置依赖设置为同一上游活动（如数据准备活动），确保它们同时启动。
3. 后续依赖：若需要等待三个并行任务全部完成再执行后续步骤，将后续活动的依赖设置为这三个活动的「成功」状态。
4. 批量管道复用：针对600条管道，使用ADF参数化功能，将表名、存储过程名设为管道参数，通过模板克隆或动态触发实现批量复用。

三、Python 脚本多线程并行方案 #

用Python结合pyodbc和多线程，直接并行调用三个子存储过程，可部署为ADF自定义活动或Azure Function：

import pyodbc
import threading

# 数据库连接字符串（替换为实际配置）
CONN_STR = (
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=YourSqlServer;"
    "DATABASE=YourDB;"
    "UID=YourUser;"
    "PWD=YourPass;"
)

def run_procedure(proc_name):
    """执行单个存储过程"""
    with pyodbc.connect(CONN_STR) as conn:
        with conn.cursor() as cursor:
            cursor.execute(f"EXEC {proc_name}")
            conn.commit()

# 定义要执行的子存储过程列表
PROC_LIST = ["proc1", "proc2", "proc3"]

# 创建并启动线程
threads = []
for proc in PROC_LIST:
    t = threading.Thread(target=run_procedure, args=(proc,))
    threads.append(t)
    t.start()

# 等待所有线程执行完成
for t in threads:
    t.join()

四、Spark 分布式并行方案 #

若原存储过程逻辑以数据复制为主，可将逻辑迁移至Spark，利用其分布式特性实现并行处理，适合超大规模数据：

方法1：多线程触发并行任务 #

from pyspark.sql import SparkSession
import threading

# 初始化SparkSession
spark = SparkSession.builder.appName("ArchiveDataParallel").getOrCreate()

# SQL Server 连接配置
SQL_SERVER_CONFIG = {
    "url": "jdbc:sqlserver://YourSqlServer:1433;databaseName=YourDB;",
    "properties": {
        "user": "YourUser",
        "password": "YourPass",
        "driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    }
}

# SQL DW 连接配置
SQL_DW_CONFIG = {
    "url": "jdbc:sqlserver://YourDwServer:1433;databaseName=YourDw;",
    "properties": {
        "user": "YourDwUser",
        "password": "YourDwPass",
        "driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    }
}

def archive_data(partition_table):
    """读取指定分区表并写入归档表"""
    df = spark.read.jdbc(
        url=SQL_SERVER_CONFIG["url"],
        table=partition_table,
        properties=SQL_SERVER_CONFIG["properties"]
    )
    df.write.jdbc(
        url=SQL_DW_CONFIG["url"],
        table=f"archive_{partition_table}",
        mode="append",
        properties=SQL_DW_CONFIG["properties"]
    )

# 对应三个子存储过程的分区表
PARTITION_TABLES = ["history_part1", "history_part2", "history_part3"]

# 并行执行归档任务
threads = [threading.Thread(target=archive_data, args=(table,)) for table in PARTITION_TABLES]
for t in threads:
    t.start()
for t in threads:
    t.join()

spark.stop()

方法2：Spark自动分区并行 #

直接将历史表按字段拆分分区，Spark自动并行读取写入，无需手动拆分任务：

df = spark.read.jdbc(
    url=SQL_SERVER_CONFIG["url"],
    table="history_table",
    column="id",  # 按主键或有序字段拆分
    lowerBound=1,
    upperBound=600000000,
    numPartitions=3,
    properties=SQL_SERVER_CONFIG["properties"]
)

df.write.jdbc(
    url=SQL_DW_CONFIG["url"],
    table="archive_table",
    mode="append",
    properties=SQL_DW_CONFIG["properties"]
)

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