使用PySpark读取列顺序不同但字段相同的多文件问题

问题根源 #

Spark批量读取带header的CSV文件时，会默认以第一个文件的列顺序作为DataFrame的固定列结构，后续文件的列会按这个位置顺序映射，完全忽略自身header的列顺序，最终导致值错位。

解决方案：逐个读取+按列名对齐合并 #

无需硬编码列顺序，核心思路是单独读取每个文件（让Spark根据各自header匹配列名与对应值），再统一调整列顺序后合并。

代码实现（Python） #

import os
from functools import reduce
from pyspark.sql import SparkSession
from pyspark.sql.functions import lit, col

spark = SparkSession.builder.appName("AlignCSVColumns").getOrCreate()

# 1. 获取所有目标CSV文件路径
csv_dir = "./"
csv_files = [
    os.path.join(csv_dir, f) 
    for f in os.listdir(csv_dir) 
    if f.endswith(".txt") and f.startswith("file")
]

# 2. 读取所有文件为独立的DataFrame
dfs = [
    spark.read.csv(file, sep=',', header=True, inferSchema=True) 
    for file in csv_files
]

# 3. 确定统一的列顺序（两种可选方案）
# 方案A：沿用第一个文件的列顺序
target_columns = dfs[0].columns
# 方案B：按列名字母排序统一顺序
# target_columns = sorted(dfs[0].columns)

# 4. 统一所有DataFrame的列顺序并合并
def align_and_union(df1, df2):
    # 处理部分文件列缺失的情况：缺失列填充null
    aligned_df2 = df2.select([
        col(c) if c in df2.columns else lit(None).alias(c) 
        for c in target_columns
    ])
    return df1.union(aligned_df2)

combined_df = reduce(align_and_union, dfs)

# 查看最终结果
combined_df.show()

原理说明 #

• 单独读取每个文件时，Spark会依据文件自身的header，将每行的值正确映射到对应的列名上，不会出现错位。
• 通过select(target_columns)将所有DataFrame的列调整为统一顺序，确保合并时列与值完全匹配。
• 额外处理列缺失场景，避免因部分文件缺少字段导致合并报错。

内容的提问来源于stack exchange，提问作者Saad Mohammad Abrar