如何用Spark/Pyspark并行查询多单租户Azure SQL数据库单表？

我正尝试将Spark用于全量刷新（full-refresh/应急）数据复制管道，以抓取数据、合并后存入分析数据库。源/原始数据仓库采用单租户（single-tenant）架构，每个Azure SQL数据库对应一个客户；目标数据仓库为多租户（multi-tenant）架构，所有客户数据存于同一数据库。

已有一个串行运行的可用示例，出于安全原因需隐藏部分代码，基本结构如下：

dfs = []
for d in databases:
    st_df = spark.read \
        .option('table', [TABLENAME]) \
        .load()
    
    dfs.append(st_df)

mt_df = reduce(lambda df1, df2: df1.unionByName(df2), dfs)

mt_df.write \
    .format([TARGET_DB]) \
    .save()

由于源数据库数量多达400余个，需要将for d in databases部分并行化，让查询并行执行以提升运行速度，该如何实现？

核心思路 #

利用Spark的分布式计算特性，将数据库列表拆分为并行任务，让多个数据库的读取操作同时执行，避免串行等待的低效问题。

方法1：基于RDD并行化读取（通用方案） #

通过将数据库配置列表转为并行RDD，在每个RDD分区中独立读取单个数据库的数据，最后合并所有结果：

from pyspark.sql import DataFrame
from functools import reduce

# 定义单个数据库的读取逻辑，需补充对应数据库的连接配置（如url、认证信息）
def read_single_database(db_config):
    return spark.read \
        .option('table', [TABLENAME]) \
        # 这里补充对应Azure SQL的连接参数，比如.jdbc(url=db_config["url"], user=..., password=...)
        .load()

# 将数据库列表并行化为RDD，numSlices建议设置为与数据库数量匹配（或根据集群资源调整）
db_rdd = spark.sparkContext.parallelize(databases, numSlices=400)

# 并行执行每个数据库的读取操作，得到包含DataFrame的RDD
df_rdd = db_rdd.map(lambda db: read_single_database(db))

# 合并所有分区的DataFrame为一个大表
mt_df = reduce(DataFrame.unionByName, df_rdd.collect())

# 写入目标多租户数据库
mt_df.write \
    .format([TARGET_DB]) \
    .save()

关键注意事项：

• numSlices需根据集群核心数、内存资源调整，避免分区过多导致资源竞争，过少无法充分并行。
• 每个数据库的连接配置要在read_single_database中正确传入，确保任务能访问到对应客户的Azure SQL实例。
• 可添加异常捕获逻辑，比如跳过读取失败的数据库或重试，避免单个失败导致整个任务终止。

方法2：使用Spark 3.0+的mapInPandas（轻量数据场景） #

如果单数据库数据量较小，可借助mapInPandas用Pandas并行读取，减少Spark与Python进程间的序列化开销：

import pandas as pd

# 批量读取分区内的数据库数据，返回Pandas DataFrame
def read_dbs_in_pandas(db_configs):
    pd_dfs = []
    for db in db_configs:
        # 用Pandas读取单个数据库，需配置好连接字符串
        pd_df = pd.read_sql([TABLENAME], con=db["connection_string"])
        pd_dfs.append(pd_df)
    return pd.concat(pd_dfs, ignore_index=True)

# 将数据库配置转为Spark DataFrame
db_df = spark.createDataFrame(databases)

# 并行处理每个分区的数据库列表，自动映射为Spark DataFrame
mt_df = db_df.mapInPandas(read_dbs_in_pandas, schema=st_df.schema)

# 写入目标库
mt_df.write \
    .format([TARGET_DB]) \
    .save()

关键注意事项：

• 需提前安装Azure SQL的Python驱动（如pyodbc），确保Pandas能正常连接数据库。
• 此方法适合单库数据量不大的场景，避免Pandas进程内存溢出。

额外优化建议 #

• 集群资源调优：增加Executor数量与核心数，确保有足够资源支撑400个并行读取任务。
• JDBC连接池：开启JDBC连接池（设置spark.sql.jdbc.maxConnections参数），减少连接创建销毁的开销。
• 失败重试：对数据库读取操作添加重试机制，处理临时网络或连接异常。
• 分区写入：写入目标库时按业务字段分区，提升写入效率与后续查询性能。

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