Azure存储中多独立Delta表的并行读取与合并性能优化问题（PySpark）

我完全懂你现在的困扰——面对一堆Schema一致但路径完全独立的Delta表，循环逐个读取再合并的效率实在太低，而且单线程循环根本没法利用Spark的分布式能力，确实让人头疼！

先帮你梳理下当前的核心困境：

• 你的Delta表都存在Azure Blob存储的独立路径下（比如wasbs://refined@mystorageaccount.blob.core.windows.net/ABC/1/、wasbs://refined@mystorageaccount.blob.core.windows.net/ABC/2/），根目录没有Delta日志，没法直接通过根路径批量读取
• 直接把路径列表传给spark.read.load()会触发Delta的报错，因为它不支持多独立路径的批量加载
• 循环读取+unionByName性能拉胯，生成器方式也只有微小提升

下面给你两个实用的优化方案，充分利用Spark的分布式能力来并行处理：

方案一：利用Spark分布式并行读取+合并 #

Spark的核心优势就是分布式计算，我们可以把路径列表并行化，让不同Executor同时读取不同路径的Delta表，再统一合并，彻底摆脱单线程循环的低效：

from pyspark.sql import DataFrame
from functools import reduce

# 定义单个路径的读取函数
def read_single_delta(path: str) -> DataFrame:
    return spark.read.load(path)

# 并行化路径列表，可根据集群资源调整numSlices（比如设置为集群Executor核心数的1-2倍）
parallel_path_rdd = sc.parallelize(paths, numSlices=len(paths))

# 并行读取所有Delta表，得到DataFrame列表
df_collection = parallel_path_rdd.map(read_single_delta).collect()

# 用reduce快速合并所有DataFrame
union_df = reduce(lambda df_a, df_b: df_a.unionByName(df_b), df_collection)

这个方法的核心是把读取任务分发到集群的多个节点并行执行，相比单循环能大幅节省读取时间。如果你的Schema完全固定且列顺序一致，也可以用union代替unionByName，性能会略高一点，但unionByName更安全，能自动匹配列名。

方案二：用SQL动态生成UNION ALL语句 #

Spark的SQL引擎会自动优化执行计划，我们可以动态拼接所有Delta表的查询语句，让SQL引擎帮我们处理并行读取和合并：

# 为每个路径生成对应的Delta表查询语句
query_fragments = [f"SELECT * FROM delta.`{path}`" for path in paths]

# 拼接成完整的UNION ALL SQL
full_union_sql = " UNION ALL ".join(query_fragments)

# 执行SQL得到合并后的DataFrame
union_df = spark.sql(full_union_sql)

这个方案的优势是无需手动管理并行逻辑，Spark SQL会自动优化读取和合并的执行流程，对于路径数量较多的场景也很友好。

额外注意事项 #

1. Schema一致性：确保所有Delta表的Schema完全一致，否则unionByName会报错。如果存在少量列差异，可以添加allowMissingColumns=True参数（比如df_a.unionByName(df_b, allowMissingColumns=True)），但要谨慎使用，避免数据丢失。
2. 并行度调整：方案一中的numSlices不要设置过大，否则会导致Executor任务过载，建议根据集群的核心数和资源情况合理设置。
3. 超大量路径处理：如果路径数量超过上千个，可以考虑分批次读取合并，避免一次性将太多DataFrame加载到Driver内存中。

备注：内容来源于stack exchange，提问作者Leonardo Bocci