PySpark读取多CSV并按列名合并方法

本文介绍如何在保证正确性（按列名对齐而非列序）的前提下，显著提升 PySpark 批量读取异构 CSV 文件的性能，避免逐文件读取的高开销，通过“分组统一读取 + 智能 schema 归类”实现接近单次加载的速度与语义正确的 unionByName 效果。

本文介绍如何在保证正确性（按列名对齐而非列序）的前提下，显著提升 PySpark 批量读取异构 CSV 文件的性能，避免逐文件读取的高开销，通过“分组统一读取 + 智能 schema 归类”实现接近单次加载的速度与语义正确的 unionByName 效果。

PySpark 原生的通配符路径读取（如 /*.csv）虽快，但其底层按列位置（index）对齐，无法处理各 CSV 文件列顺序不同、列集不全等常见异构场景——正如示例中 2.csv 缺失列 B 且 C 位于第二列，直接合并会导致数据错位（8 被错误填入 B 列）。而逐文件读取 + unionByName(allowMissingColumns=True) 虽语义正确，却因多次 Spark 作业调度、重复解析开销，性能下降明显（100 文件从 6s 延至 16s）。

真正的高效解法在于减少读取次数 + 保留列名语义，核心思路是：先轻量获取所有文件的 header（首行），按 schema 结构聚类；再对每组结构一致的文件批量读取，最后组间 unionByName。

✅ 推荐优化方案（两阶段高性能流程）

第一阶段：Schema 分组（纯 Python，极快）

无需 Spark，仅用轻量 I/O 获取每个 CSV 的列名列表，并哈希归类：

from collections import defaultdict
import os

def get_csv_headers(file_path):
    """安全读取 CSV 首行（跳过空行/BOM），返回列名元组"""
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip():
                return tuple(col.strip() for col in line.strip().split(','))
    return tuple()

# 示例：本地目录分组（HDFS/S3 需替换为对应 SDK，如 boto3 或 Hadoop FS）
base_dir = "/path/to/csv/folder"
schema_groups = defaultdict(list)

for fname in os.listdir(base_dir):
    if fname.endswith(".csv"):
        full_path = os.path.join(base_dir, fname)
        try:
            header = get_csv_headers(full_path)
            schema_groups[header].append(full_path)
        except Exception as e:
            print(f"Skip {fname}: {e}")

# 输出分组结果（例如）：
# {('A','B','C'): ['1.csv'], ('A','C'): ['2.csv', '3.csv']}

⚠️ 注意：若文件在 HDFS 或 S3，需改用 hdfs.client.Client 或 boto3.S3Client.get_object() 流式读取前几 KB 获取首行，避免下载全量文件。

第二阶段：分组批量读取 + 合并

对每个 schema 组使用通配符一次性读取（保留高效性），再跨组 unionByName：

from pyspark.sql import DataFrame

dfs_by_schema = []
for schema, paths in schema_groups.items():
    # 构造路径字符串（Spark 支持逗号分隔多路径）
    path_list = ",".join(paths)
    df = spark.read.format("csv") \
        .option("header", "true") \
        .option("inferSchema", "false") \  # 关闭推断，提速且确保列序与 header 严格一致
        .load(path_list)
    dfs_by_schema.append(df)

# 按 schema 复杂度排序（可选：让宽表优先，减少中间 shuffle）
dfs_by_schema.sort(key=lambda df: len(df.columns), reverse=True)

# 逐组 unionByName（自动对齐列名，缺失列补 null）
result_df = dfs_by_schema[0]
for df in dfs_by_schema[1:]:
    result_df = result_df.unionByName(df, allowMissingColumns=True)

✅ 性能对比与关键优势

• 为什么更快？
  • 减少 Spark 任务启动开销（从 100 次降至 2–5 次）；
  • 每组内利用 Spark 原生 CSV 批处理优化（向量化解析、内存复用）；
  • inferSchema=False 避免重复类型推断，进一步提速。

? 补充建议

• 动态列处理：若列名存在大小写/空格差异，预处理 header 时统一标准化（如 col.lower().strip()）。
• 元数据缓存：将 schema_groups 结果持久化（如 JSON 文件），后续增量更新只需比对新增文件。
• 云存储适配：S3 上使用 s3a://bucket/path/*.csv 通配符本身支持，但 header 探测需 boto3；HDFS 可用 hadoop fs -cat 管道流式提取。

该方案在保持 unionByName 语义严谨性的前提下，逼近原生通配符读取的性能边界，是生产环境中处理异构 CSV 批量加载的最佳实践路径。