如何并行读取Schema各异的Part文件夹并对齐预定义Schema？

问题场景 #

存在一批按日期划分的Parquet分区文件夹（示例路径如下），每个分区的Schema可能存在列数不一致、部分列数据类型不匹配的情况。需求是读取所有分区数据，最终生成符合预定义Schema的单一DataFrame。

示例路径结构：

/feed=abc -> 包含多个日期分区文件夹：
/feed=abc/date=20221220
/feed=abc/date=20221221
.....
/feed=abc/date=20221231

当前采用串行处理逻辑：逐个读取分区文件夹，与预定义Schema对比后执行列增删、类型转换，将修正后的数据写入临时目录；待所有分区处理完成后，读取临时目录合并得到最终结果。现希望改为并行处理（通过线程/进程同步处理多个分区），询问该方案是否可行，且分区数量可能超过1000，常规通配符读取（适用于Schema一致场景）无法满足需求。

方案可行性结论 #

该方案完全可行，但需结合Spark的分布式特性设计实现，避免本地并行带来的资源冲突问题。以下是具体实现思路与注意事项：

具体实现思路 #

1. 基于Spark分布式并行处理（推荐） #

利用Spark原生的分布式计算能力实现并行处理，无需依赖本地线程/进程，更适合超1000分区的大规模场景：

• 收集分区路径：通过Hadoop FileSystem API遍历所有目标分区目录，将路径列表转换为Spark Dataset[String]
• 并行处理每个分区：使用mapPartitions或foreachPartition对路径数据集进行分布式处理，每个分区任务独立完成：
  1. 读取当前路径下的Parquet文件，生成原始DataFrame
  2. 调用Schema对齐函数，完成列增删、类型转换（逻辑见下文）
  3. 将修正后的DataFrame写入临时目录的独立子目录（以分区日期命名，避免文件冲突）
• 合并结果：所有分区处理完成后，读取临时目录下的所有数据，合并为符合预定义Schema的单一DataFrame

2. Schema对齐通用函数 #

封装统一的Schema修正逻辑，确保每个分区数据都能对齐到目标Schema：

from pyspark.sql import DataFrame
from pyspark.sql.types import StructType
from pyspark.sql.functions import lit, try_cast

def align_schema(df: DataFrame, target_schema: StructType) -> DataFrame:
    # 构建当前列与目标列的类型映射
    current_col_types = {field.name: field.dataType for field in df.schema.fields}
    target_col_types = {field.name: field.dataType for field in target_schema.fields}

    # 保留目标Schema中存在的列
    filtered_df = df.select(*[col for col in df.columns if col in target_col_types])

    # 补全缺失列并转换类型（用try_cast避免转换失败导致任务崩溃）
    for col_name, target_type in target_col_types.items():
        if col_name not in current_col_types:
            filtered_df = filtered_df.withColumn(col_name, lit(None).cast(target_type))
        else:
            filtered_df = filtered_df.withColumn(col_name, try_cast(filtered_df[col_name], target_type))

    # 按目标Schema的列顺序重新排列
    return filtered_df.select(*target_schema.fieldNames())

3. 临时目录管理 #

• 为每个分区分配独立的临时子目录（例如/tmp/aligned_data/date=20221220），避免并行写入时的文件冲突
• 处理完成后可通过Spark API或文件系统命令自动清理临时目录（若无需保留中间数据）

关键注意事项 #

• 资源并行度控制：针对超1000分区的场景，需调整Spark参数（如spark.sql.shuffle.partitions）控制并行任务数量，避免集群资源耗尽
• 异常处理：使用try_cast替代原生cast处理类型转换，避免非法值导致任务失败；可添加日志记录转换失败的行或分区
• 路径遍历效率：使用Hadoop FileSystem API批量获取分区路径，比手动列举更高效且不易遗漏
• 原子性保障：每个分区的写入操作需确保原子性，可采用Spark的mode("overwrite")写入独立子目录，避免部分写入导致的数据不一致

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