PySpark中Schema重命名致聚合异常及写入失败问题排查

我有一个字段名含无效字符（点和括号）的大型Parquet DataFrame。直接读取并执行聚合操作可得到正确结果：

>>> df1 = spark.read.parquet(source)
>>> df1.select('`A.B.C`').fillna(0).agg({'`A.B.C`' : 'max'}).show()
+----------+
|max(A.B.C)|
+----------+
|  5989.625|
+----------+

为便于后续操作，我尝试重命名列，遇到两个问题：

1. 自定义Schema读取重命名后聚合结果异常

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)

聚合结果：

>>> df2.select('A__B__C').fillna(0).agg({'A__B__C' : 'max'}).show()
+------------+
|max(A__B__C)|
+------------+
|         0.0|
+------------+

2. select别名重命名后聚合正常但写入失败

mapping = {c : c.replace('.', '___').replace('(', '-').replace(')', '-') for c in df1.columns}
df3 = df1.select([F.col(f'`{c}`').alias(mapping[c]) for c in df1.columns])

聚合结果正常：

>>> df3.select('A__B__C').fillna(0).agg({'A__B__C' : 'max'}).show()
+------------+
|max(A__B__C)|
+------------+
|    5989.625|
+------------+

但写入表时报错：Attribute name "Some.Col(1)" contains invalid character(s) among " ,;{}()\n\t="，尽管"Some.Col(1)" in df3.columns返回False，"Some__Col-1-" in df3.columns返回True。

另外，withColumnRenamed逐个重命名耗时过长，未测试。

• 为何自定义Schema重命名会导致聚合结果异常？
• 为何select别名重命名聚合正常但写入失败？
• 如何正确实现列重命名？

1. 自定义Schema重命名导致聚合异常的原因 #

Parquet文件存储时，列匹配是按原始字段的元数据名称/路径进行的，而非字段位置。当你修改Schema中的列名后，Spark会用新列名去匹配Parquet文件里的原始列名，匹配失败的列会被填充为null。后续fillna(0)将这些空值替换为0，最终聚合得到的最大值自然是0。

简单来说：你改了Schema的列名，但Parquet文件里的列还是A.B.C，Spark找不到A__B__C对应的字段，所以该列全为null，max结果为0。

2. select别名重命名后写入失败的原因 #

虽然alias修改了DataFrame表面的列名，但Spark的逻辑计划中可能仍保留了原始列名的元数据（比如Parquet字段的原始名称信息）。当写入表时，底层校验会读取这些隐藏的元数据，发现原始含无效字符的列名，从而抛出错误。

如果写入的是Hive表，Hive元数据对列名格式要求更严格，即使DataFrame显示的列名合法，底层携带的原始字段元数据也会触发校验失败。

3. 正确的列重命名方案 #

方案一：读取后批量重命名（高效替代withColumnRenamed） #

通过toDF一次性重命名所有列，性能远优于循环调用withColumnRenamed：

# 读取原始DataFrame
df = spark.read.parquet(source)
# 生成新列名列表
new_columns = [c.replace('.', '___').replace('(', '-').replace(')', '-') for c in df.columns]
# 一次性完成重命名
df_renamed = df.toDF(*new_columns)

这个方法直接生成新的列名映射，不会保留原始列名的元数据，写入时不会触发无效字符校验。

方案二：读取时按位置匹配Schema（解决自定义Schema的匹配问题） #

如果要在读取阶段处理，需确保新Schema的字段顺序、类型、nullable属性完全匹配原始Parquet文件，然后强制按位置匹配：

# 获取原始Schema的字段结构
original_fields = df1.schema.fields
# 生成新字段（仅修改名称，保留其他属性）
new_fields = [
    StructField(
        f.name.replace('.', '___').replace('(', '-').replace(')', '-'),
        f.dataType,
        f.nullable
    ) for f in original_fields
]
new_schema = StructType(new_fields)
# 读取时指定Schema，并关闭mergeSchema强制按位置匹配
df2 = spark.read.schema(new_schema).option("mergeSchema", "false").parquet(source)

注意：此方法要求原始Parquet文件的字段顺序与新Schema完全一致，否则会出现字段错位。

验证方式 #

重命名后可通过以下方式确认列名和元数据是否正确：

# 查看Schema确认列名
df_renamed.printSchema()
# 查看逻辑计划，确认无原始列名残留
print(df_renamed.queryExecution.logical)

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