如何用PySpark读取列数不同的多个CSV并合并为单DataFrame

直接使用spark.read.csv(Files, header=True)读取多列数不同的CSV时，Spark会自动以列数最少的文件结构作为最终Schema，导致缺失列无法被读取。要实现类似Pandasconcat的效果（保留所有列，缺失值填充NaN），可以按以下步骤操作：

1. 收集所有文件的完整列名集合 #

先遍历所有CSV文件，提取每个文件的表头，汇总得到所有列名的全集：

方式一：纯PySpark实现 #

from pyspark.sql import SparkSession

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

Files = ['Data/f1.csv','Data/f2.csv','Data/f3.csv','Data/f4.csv','Data/f5.csv']

all_columns = set()
for file in Files:
    # 读取文件第一行（表头）
    header_row = spark.read.text(file).limit(1).first()[0]
    # 按CSV分隔符拆分列名（默认逗号，若为其他分隔符需修改）
    cols = header_row.split(',')
    all_columns.update(cols)
# 转为有序列表（可选，保证列顺序统一）
all_columns = sorted(all_columns)

方式二：借助Pandas快速提取表头（小文件更高效） #

import pandas as pd

all_columns = set()
for file in Files:
    # 仅读取表头，不加载数据
    temp_df = pd.read_csv(file, nrows=0)
    all_columns.update(temp_df.columns)
all_columns = sorted(all_columns)

2. 逐个读取文件并补全缺失列 #

对每个CSV文件单独读取，检查并添加缺失的列（用lit(None)填充NaN），最后统一列顺序：

from pyspark.sql.functions import lit

df_list = []
for file in Files:
    # 读取单个CSV文件
    df = spark.read.csv(file, header=True, inferSchema=True)
    # 补全缺失列
    for col in all_columns:
        if col not in df.columns:
            df = df.withColumn(col, lit(None))
    # 统一列顺序
    df = df.select(all_columns)
    df_list.append(df)

# 按列名合并所有DataFrame（unionByName会自动匹配列名，无需担心顺序）
final_df = df_list[0]
for df in df_list[1:]:
    final_df = final_df.unionByName(df)

3. 关键注意事项 #

• 分隔符适配：如果CSV使用非逗号分隔符（如制表符），需在spark.read.csv中指定sep参数（例如sep='\t'）。
• 数据类型一致性：inferSchema可能导致同一列在不同文件中被推断为不同类型，若需严格统一类型，可先定义全局Schema（例如所有列先设为StringType，后续再按需转换）。
• 性能优化：若文件数量极大，可抽样读取部分文件获取列名；对于5个文件的场景，直接遍历完全可行。

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