如何强制PySpark读取Parquet文件时使用指定Schema而非自动推断

现有一个包含无效字符的Parquet文件，需要通过自定义Schema重命名字段（新旧Schema字段数量、顺序完全一致，仅字段名修改）。执行以下代码后，仅新旧Schema共有的列能正常读取，新Schema独有的列（即重命名后的列）全部显示为null，修改mergeSchema选项也无效：

df1 = spark.read.parquet(source)
schema = df1.schema
new_fields = [
    StructField(f.name.replace('.', '___').replace('(', '-').replace(')', '-'), f.dataType, f.nullable) for f in schema.fields
]
new_schema = StructType(new_fields)

df2 = spark.read.schema(new_schema).parquet(source)

Spark的Parquet读取器默认按列名匹配，而非字段顺序或位置匹配，所以直接指定新Schema读取时，重命名后的字段找不到对应原始列，会返回null。正确的做法是先读取原始数据，再批量重命名列：

1. 读取原始Parquet文件得到DataFrame
2. 遍历原始列和新字段名，对列进行批量重命名
3. （可选）验证最终DataFrame的Schema是否符合预期

修改后的代码示例：

from pyspark.sql.types import StructField, StructType

# 读取原始数据
df1 = spark.read.parquet(source)
original_schema = df1.schema

# 生成新字段名列表
new_field_names = [
    f.name.replace('.', '___').replace('(', '-').replace(')', '-') 
    for f in original_schema.fields
]

# 批量重命名列
df2 = df1.toDF(*new_field_names)

# 验证Schema（可选）
print(df2.schema)

如果需要严格对齐自定义的new_schema（确保字段顺序、类型完全一致），可在重命名后再做转换：

# 生成完整的新Schema
new_fields = [
    StructField(name, f.dataType, f.nullable) 
    for name, f in zip(new_field_names, original_schema.fields)
]
new_schema = StructType(new_fields)

# 转换为目标Schema（可选，重命名后已基本匹配）
df2 = df2.select([df2[col].alias(new_schema.fields[i].name) for i, col in enumerate(df2.columns)]).cast(new_schema)

关键说明 #

• 直接指定schema=new_schema读取时，Spark会在Parquet元数据中寻找与新字段名完全匹配的列，找不到则返回null
• 通过toDF(*new_names)或逐个重命名列的方式，是基于原始列的位置/索引映射，能确保所有列都被正确重命名

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