PySpark实现含ADDRESS字段列的批量重命名通用方案

以下是可直接嵌入数据处理流水线的PySpark代码，能自动识别所有名称包含ADDRESS的列——不管列名带不带前缀（比如T_）——统一重命名为S_ADDRESS1、S_ADDRESS2、S_ADDRESS3这类规范格式：

from pyspark.sql import SparkSession
import re

# 初始化SparkSession（流水线中已存在则跳过）
spark = SparkSession.builder.appName("AddressStandardization").getOrCreate()

# 读取数据源（示例为CSV，可替换为Parquet、JSON等你实际的数据源格式）
df = spark.read.csv("your_input_path", header=True, inferSchema=True)

# 定义列名标准化逻辑
def standardize_address_col(col_name):
    # 正则匹配提取ADDRESS后的数字，兼容大小写、带前缀/不带前缀的情况
    match = re.search(r'ADDRESS(\d+)', col_name.upper())
    if match:
        num = match.group(1)
        return f"S_ADDRESS{num}"
    # 非ADDRESS相关列保持原名
    return col_name

# 应用重命名规则
renamed_df = df.toDF(*[standardize_address_col(col) for col in df.columns])

# 验证结果（流水线中可注释，仅调试用）
renamed_df.printSchema()

# 保存处理后的数据（按需调整输出格式和路径）
renamed_df.write.csv("your_output_path", header=True, mode="overwrite")

关键逻辑说明 #

• 正则r'ADDRESS(\d+)'会忽略列名的大小写和前缀，精准捕获ADDRESS后的数字部分，不管是ADDRESS1、T_ADDRESS2还是TADDRESS3都能匹配。
• 遍历所有列仅修改符合规则的地址列，不会改动其他业务列，适配多文件不同列名格式的场景。
• 代码可直接对接现有流水线，只需替换数据源读取和输出的对应方法即可。

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