如何基于另一DataFrame动态修改PySpark DataFrame的列名

我来帮你搞定这个动态列名替换的需求！核心思路是先把映射表转换成可快速查询的字典，再批量完成列名重命名，完全不用硬编码，不管映射表有多少行都能自动适配。

步骤拆解： #

• 生成列名映射字典：从df_col中提取col_current和col_updated的对应关系，同时处理掉字符串里的多余空格（看你的示例数据，映射列里存在前导空格，不处理会导致匹配失败）。
• 批量重命名列：遍历原始DataFrame的所有列，用映射字典匹配新列名，批量生成重命名表达式后，通过select完成转换。

完整代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# 初始化SparkSession
spark = SparkSession.builder.appName("DynamicRenameColumns").getOrCreate()

# 创建测试用的原始数据df
data_df = [("ASD", 101, "John", "DEV"), ("klj", 102, "ben", "prod")]
df = spark.createDataFrame(data_df, ["code", "id", "name", "work"])

# 创建测试用的列名映射表df_col
data_col = [(" id", "Row_id"), (" name", "Name"), (" code", "Row_code"), (" Work", "Work_Code")]
df_col = spark.createDataFrame(data_col, ["col_current", "col_updated"])

# 1. 生成列名映射字典，处理字符串空格
col_mapping = {
    row.col_current.strip(): row.col_updated.strip() 
    for row in df_col.collect()
}

# 2. 批量生成重命名后的列表达式
renamed_columns = [
    col(original_col).alias(col_mapping.get(original_col.strip(), original_col.strip()))
    for original_col in df.columns
]

# 3. 生成最终的重命名后DataFrame
df_renamed = df.select(*renamed_columns)

# 查看结果
df_renamed.show()

输出结果 #

+-------+----+---------+----------+
|Row_id |Name|Row_code |Work_Code |
+-------+----+---------+----------+
|101    |John|ASD      |DEV       |
|102    |ben |klj      |prod      |
+-------+----+---------+----------+

方案优势： #

• 完全动态：不管df_col新增多少映射规则，代码无需修改，自动适配。
• 鲁棒性强：通过strip()处理字符串空格，避免因输入数据格式问题导致匹配失败。
• 效率更高：用select批量处理列，比循环调用withColumnRenamed更高效（尤其是列数较多时）。

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