基于Python并行处理的19亿条SQL Server到BigQuery数据迁移优化

一、Python并行处理提速方案 #

你的现有脚本采用串行逻辑，每块数据的读取与写入需等待完成后才进行下一块，这是耗时的核心原因。可通过多进程分区间并行处理提升效率，同时优化数据读写方式。

1. 多进程分区间并行处理 #

先获取ID的范围，将数据拆分为多个独立区间，每个进程单独处理一个区间的读取与写入，避免重复读取，同时利用多核CPU资源。注意每个进程需单独创建数据库连接（pyodbc连接非线程安全）。

import pandas as pd
import pyodbc
from google.cloud import bigquery
from concurrent.futures import ProcessPoolExecutor

# 配置参数请自行替换
SERVER = "your_sql_server"
DATABASE = "your_db"
USERNAME = "your_user"
PASSWORD = "your_pwd"
SOURCE_TABLE = "source_table"
BQ_TABLE_REF = "your_project.your_dataset.target_table"
CHUNK_SIZE = 1000000  # 每块数据量

def get_id_boundaries():
    """获取SQL Server表中ID的最小/最大值"""
    conn_str = f'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={SERVER};DATABASE={DATABASE};UID={USERNAME};PWD={PASSWORD}'
    with pyodbc.connect(conn_str) as conn:
        min_id = pd.read_sql(f"SELECT MIN(id) FROM {SOURCE_TABLE}", conn).iloc[0, 0]
        max_id = pd.read_sql(f"SELECT MAX(id) FROM {SOURCE_TABLE}", conn).iloc[0, 0]
    return min_id, max_id

def process_chunk(start_id, end_id):
    """处理单个ID区间的数据迁移"""
    # 每个进程单独创建数据库连接和BigQuery客户端
    conn_str = f'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={SERVER};DATABASE={DATABASE};UID={USERNAME};PWD={PASSWORD}'
    conn = pyodbc.connect(conn_str)
    bq_client = bigquery.Client()

    # 配置BigQuery加载任务
    job_config = bigquery.LoadJobConfig(
        autodetect=False,  # 建议手动指定schema，避免自动检测开销
        write_disposition="WRITE_APPEND",
    )

    # 查询区间内的数据（ID有索引时无需ORDER BY，可快速定位）
    query = f"SELECT * FROM {SOURCE_TABLE} WHERE id > {start_id} AND id <= {end_id}"
    df = pd.read_sql(query, conn)

    if not df.empty:
        load_job = bq_client.load_table_from_dataframe(df, BQ_TABLE_REF, job_config=job_config)
        load_job.result()  # 等待加载完成
        print(f"完成区间迁移: {start_id} -> {end_id}")

    conn.close()

if __name__ == "__main__":
    min_id, max_id = get_id_boundaries()
    # 生成所有待处理的ID区间
    chunks = []
    current_start = min_id
    while current_start < max_id:
        current_end = min(current_start + CHUNK_SIZE, max_id)
        chunks.append((current_start, current_end))
        current_start = current_end

    # 启动多进程处理，max_workers根据CPU核心数和数据库负载调整（建议4-8）
    with ProcessPoolExecutor(max_workers=6) as executor:
        executor.map(process_chunk, [c[0] for c in chunks], [c[1] for c in chunks])

2. 优化数据读写性能 #

• 替换pandas read_sql为Parquet导出：将数据导出为Parquet列存格式，再上传到BigQuery，相比直接写入DataFrame，内存开销更小、加载速度更快。
• 关闭autodetect：手动定义BigQuery表的schema，避免自动检测带来的额外耗时。

二、非Python的高效迁移方案 #

对于19亿条数据的超大规模迁移，原生工具或托管式ETL框架通常比Python脚本效率更高：

1. SQL Server bcp工具 + GCS + BigQuery批量加载 #

• 用SQL Server原生bcp工具导出数据为Parquet/CSV，速度远快于Python读取：

  bcp [数据库名].[架构名].[表名] out 本地文件路径.parquet -S [SQL Server地址] -U [用户名] -P [密码] -d [数据库名] -c -t ","
  

• 用gsutil将文件上传到Google Cloud Storage：

  gsutil cp 本地文件路径.parquet gs://你的存储桶名/
  

• 在BigQuery中执行批量加载：

  LOAD DATA INPATH 'gs://你的存储桶名/文件名.parquet' 
  INTO TABLE [项目ID].[数据集ID].[目标表名] 
  FORMAT PARQUET;
  

2. Google Cloud Dataflow托管迁移 #

使用Dataflow的预定义模板（SQL Server to BigQuery），无需编写代码即可实现分布式并行迁移：

• 配置SQL Server连接信息、BigQuery目标表参数
• Dataflow自动调度分布式资源处理数据，适合超大规模数据迁移，且无需维护并行逻辑

3. 分阶段迁移 #

• 历史数据：用bcp+GCS+BigQuery批量加载，快速完成大规模数据迁移
• 增量数据：用Python脚本或Dataflow实时同步新增数据，减少单次迁移压力

三、现有脚本的快速优化点 #

• 移除ORDER BY：WHERE id > last_id结合ID索引，无需排序即可快速获取下一批数据，排序会增加SQL Server查询耗时
• 增大分块大小：将100万调整为500万甚至1000万，减少作业调度的开销
• 复用BigQuery客户端：串行脚本中避免每次循环创建客户端，提升效率

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