合并表头顺序不同的多个CSV文件（PySpark实现问题）

我之前处理多CSV合并的时候也碰到过一模一样的问题——PySpark默认会按照每个文件自身的列顺序来加载数据，直接union的话自然就乱套了。下面分享几个我亲测好用的解决方法，你可以根据自己的场景选：

方法1：指定统一目标列顺序（最直接的方案） #

先确定一个你想要的标准列顺序（比如从第一个文件里提取，或者手动写死），然后每个CSV读取后都强制按这个顺序重排列，再合并就不会乱了。

示例代码：

from pyspark.sql import SparkSession

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

# 定义你想要的最终列顺序（可以从基准文件读取，也可以手动指定）
target_columns = ["user_id", "name", "age", "register_time"]

# 遍历所有CSV文件，逐个调整列顺序
csv_files = ["data1.csv", "data2.csv", "data3.csv"]
df_list = []
for file_path in csv_files:
    # 读取单个CSV
    raw_df = spark.read.csv(file_path, header=True, inferSchema=True)
    # 按目标列顺序重排（确保列名完全匹配，这里假设所有文件都包含这些列）
    reordered_df = raw_df.select(target_columns)
    df_list.append(reordered_df)

# 合并所有DataFrame
merged_df = df_list[0].unionAll(*df_list[1:])

# 保存结果
merged_df.write.csv("merged_output.csv", header=True, mode="overwrite")

如果担心某个文件可能缺失部分列，可以加个判断逻辑，自动补全null：

from pyspark.sql.functions import lit

reordered_cols = []
for col_name in target_columns:
    if col_name in raw_df.columns:
        reordered_cols.append(raw_df[col_name])
    else:
        # 缺失的列填充null
        reordered_cols.append(lit(None).alias(col_name))
reordered_df = raw_df.select(reordered_cols)

方法2：读取时指定统一Schema（高效稳定，适合大文件） #

如果你提前知道所有列的名称和数据类型，直接定义一个标准Schema，让Spark读取所有文件时都按这个Schema来加载——这样不仅能自动对齐列顺序，还能避免Spark重复推断Schema（大文件下inferSchema会很慢）。

示例代码：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType

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

# 定义标准Schema，列顺序就是你想要的最终顺序
standard_schema = StructType([
    StructField("user_id", IntegerType(), nullable=True),
    StructField("name", StringType(), nullable=True),
    StructField("age", IntegerType(), nullable=True),
    StructField("register_time", TimestampType(), nullable=True)
])

csv_files = ["data1.csv", "data2.csv", "data3.csv"]
df_list = []
for file_path in csv_files:
    # 读取时指定Schema，自动对齐列顺序
    df = spark.read.csv(file_path, header=True, schema=standard_schema)
    df_list.append(df)

merged_df = df_list[0].unionAll(*df_list[1:])
merged_df.write.csv("merged_output.csv", header=True, mode="overwrite")

要是不确定Schema，也可以先读一个基准文件获取Schema，再用它来读其他文件：

# 从第一个文件获取Schema作为标准
base_df = spark.read.csv(csv_files[0], header=True, inferSchema=True)
standard_schema = base_df.schema

# 用这个Schema读取剩余文件
df_list = [base_df]
for file_path in csv_files[1:]:
    df = spark.read.csv(file_path, header=True, schema=standard_schema)
    df_list.append(df)

方法3：动态对齐列顺序（灵活处理列名变动） #

如果你的文件列名可能有小变动（比如部分文件多几列或少几列），可以先收集所有文件的列名交集，再按统一顺序排序后重排每个DataFrame。

示例代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

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

csv_files = ["data1.csv", "data2.csv", "data3.csv"]

# 先收集所有文件的列名
all_columns = set()
for file_path in csv_files:
    temp_df = spark.read.csv(file_path, header=True, inferSchema=True)
    all_columns.update(temp_df.columns)

# 以第一个文件的列顺序为基准，补充其他列（按字母排序）
base_df = spark.read.csv(csv_files[0], header=True, inferSchema=True)
target_columns = [col for col in base_df.columns if col in all_columns]
extra_columns = sorted([col for col in all_columns if col not in target_columns])
target_columns += extra_columns

# 遍历文件，调整每个DataFrame的列顺序
df_list = []
for file_path in csv_files:
    raw_df = spark.read.csv(file_path, header=True, inferSchema=True)
    # 生成要选择的列，缺失的列补null
    selected_cols = []
    for col_name in target_columns:
        if col_name in raw_df.columns:
            selected_cols.append(raw_df[col_name])
        else:
            selected_cols.append(lit(None).alias(col_name))
    reordered_df = raw_df.select(selected_cols)
    df_list.append(reordered_df)

merged_df = df_list[0].unionAll(*df_list[1:])
merged_df.write.csv("merged_output.csv", header=True, mode="overwrite")

总结 #

• 小文件、列名固定：用方法1，简单直接
• 大文件、列类型明确：用方法2，高效不卡
• 列名可能变动：用方法3，灵活容错

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