PySpark批量移除DataFrame列名特殊字符报错：无法解析指定列

问题根源 #

你遇到的报错核心原因有两个：

• 当列名包含空格、横杠、点号等特殊字符时，直接用F.col(列名字符串)会导致Spark解析失败：它会把特殊字符当成标识符分隔符，比如Organization - No. Of Employees会被解析成取Organization列下的No子字段，自然找不到对应列。
• 你做了两次不必要的列名修改：第一次用正则替换后df的列名已经更新，再用原始列名索引新df也会出现列不存在的问题。

解决代码 #

步骤1：导入依赖 #

import re
from pyspark.sql import functions as F

步骤2：读取CSV（原有读取逻辑可以保留） #

df = spark.read.format("com.databricks.spark.csv") \
  .option("mode", "DROPMALFORMED") \
  .option("header", "true") \
  .option("inferschema", "true") \
  .option("delimiter", ",").load(getArgument('sourceCSVpath') + getArgument('sourceCSV'))

步骤3：统一清洗列名 #

定义通用清洗函数，同时用反引号包裹原始列名避免解析错误：

def clean_column_name(col_name, col_index):
    # 替换所有非字母、数字、下划线的字符，可根据需求调整允许保留的字符
    cleaned = re.sub(r'[^0-9a-zA-Z_]+', '', col_name)
    # 兼容数字开头的列名、空列名的异常场景
    if not cleaned:
        return f"col_{col_index}"
    if cleaned[0].isdigit():
        return f"c_{cleaned}"
    return cleaned

# 一次性完成列名替换，不需要多次select
df_clean = df.select(
    [F.col(f"`{old_col}`").alias(clean_column_name(old_col, idx)) for idx, old_col in enumerate(df.columns)]
)

步骤4：验证列名映射（可选，方便排查） #

for old, new in zip(df.columns, df_clean.columns):
    print(f"{old} -> {new}")

注意事项 #

• 正则规则可自行调整，比如要保留$字符，就把正则改成r'[^0-9a-zA-Z_$]+'即可。
• 如果存在清洗后列名重复的情况，可以在清洗函数中加去重逻辑，给重复列名添加序号后缀即可。

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