Spark读取S3中Schema不一致的CSV，如何解决列偏移问题？

你的问题核心是部分CSV文件存在非末尾的额外列，Spark默认采样推断的Schema不包含这些列，读取时按位置映射导致列偏移，且无法手动指定Schema。下面是可以实现扫描所有文件确定统一Schema的方案：

方案：先收集所有文件表头生成统一Schema，再读取数据 #

这个方法会先遍历所有CSV文件的表头，收集所有出现过的列名，生成包含所有列的Schema，最后用这个Schema读取数据，Spark会根据表头匹配列而非位置，彻底避免偏移问题。

步骤及代码示例 #

1. 读取所有文件的表头行
   用wholeTextFiles读取所有CSV文件，提取每一个文件的第一行（表头）：

# 读取目标路径下的所有CSV文件，提取每行的表头
header_rdd = self.spark.sparkContext.wholeTextFiles(self.input_loc) \
    .map(lambda file_content: file_content[1].split('\n')[0].strip())

2. 收集所有唯一列名
   遍历所有表头，合并得到所有出现过的列名，排序保证列顺序统一：

all_columns = set()
# 遍历所有表头，收集列名
for header in header_rdd.collect():
    columns = [col.strip() for col in header.split(',')]
    all_columns.update(columns)
# 排序确保列顺序一致，避免后续读取出现列乱序
all_columns = sorted(all_columns)

3. 生成统一Schema
   基于收集到的所有列名创建Schema，所有列先设为String类型（后续可按需调整类型）：

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

unified_schema = StructType([
    StructField(col_name, StringType(), nullable=True) 
    for col_name in all_columns
])

4. 用统一Schema读取所有CSV文件
   读取时指定生成的统一Schema，同时保持你原有的配置：

df = self.spark.read.csv(
    self.input_loc,
    sep=",",
    header="true",
    schema=unified_schema,
    ignoreLeadingWhiteSpace="true",
    ignoreTrailingWhiteSpace="true",
    mode="PERMISSIVE"  # 保留所有行，缺失列自动填null；若要丢弃格式错误行可改用DROPMALFORMED
)

注意事项 #

• 当文件数量极大（比如单分区2万+），collect()可能会占用较多Driver内存，可改用reduce操作合并列名，避免一次性加载所有表头到Driver：

  def merge_columns(cols_set, header):
      columns = [col.strip() for col in header.split(',')]
      cols_set.update(columns)
      return cols_set
  
  all_columns = header_rdd.reduce(merge_columns, set())
  all_columns = sorted(all_columns)
  

• 若后续需要推断列类型，可对读取后的DataFrame使用inferSchema，或根据业务逻辑手动指定各列类型；
• mode="PERMISSIVE"是默认值，会把表头和值数量不匹配的行放到_corrupt_record列，可根据需求选择其他模式（比如DROPMALFORMED直接丢弃错误行）。

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