如何在PySpark中处理读取CSV文件时的缺失列问题

方案1：显式定义目标Schema（优先推荐） #

你已经明确需要五列String类型的固定结构，直接提前定义好Schema即可，Spark读取时会自动为缺失列填充null，无需额外做缺失检测。

from pyspark.sql.types import StructType, StructField, StringType

# 定义预期的完整表结构
target_schema = StructType([
    StructField("A", StringType(), nullable=True),
    StructField("B", StringType(), nullable=True),
    StructField("C", StringType(), nullable=True),
    StructField("D", StringType(), nullable=True),
    StructField("E", StringType(), nullable=True)
])

# 批量读取所有CSV文件
df = spark.read.csv(
    path="/your/csv/path/*.csv",
    schema=target_schema,
    header=True, # CSV带表头则设为True，无表头设为False
    encoding="utf-8"
)

注意：如果CSV无表头，要保证所有文件现有列的顺序和Schema定义的顺序完全一致，否则会出现字段错位问题。

方案2：动态合并Schema（适用字段不固定的场景） #

如果无法提前确定完整字段列表，可以先遍历所有文件收集Schema合并，再补全缺失列后合并数据：

import os
from pyspark.sql.functions import lit
from pyspark.sql.types import StringType

csv_dir = "/your/csv/path/"
csv_paths = [os.path.join(csv_dir, f) for f in os.listdir(csv_dir) if f.endswith(".csv")]

# 收集所有文件的字段得到完整字段集合
all_cols = set()
for p in csv_paths:
    temp_df = spark.read.csv(p, header=True, nrows=1)
    all_cols.update(temp_df.columns)
# 按需要的顺序调整字段排序
sorted_cols = sorted(all_cols, key=lambda x: ["A","B","C","D","E"].index(x))

# 逐个读取文件补全缺失列
df_list = []
for p in csv_paths:
    temp_df = spark.read.csv(p, header=True, encoding="utf-8")
    for col in sorted_cols:
        if col not in temp_df.columns:
            temp_df = temp_df.withColumn(col, lit(None).cast(StringType()))
    temp_df = temp_df.select(sorted_cols)
    df_list.append(temp_df)

# 合并所有数据
final_df = df_list[0]
for df in df_list[1:]:
    final_df = final_df.unionByName(df)

该方案兼容性更强，即使后续出现其他列缺失的情况也能自动适配，缺点是需要两次遍历文件，大数据量下会有额外性能开销。

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