PySpark如何批量重命名嵌套结构列名中的点符号？

针对多层嵌套（struct+array）的JSON数据，无需硬编码Schema，基于自动推断结构重命名含特殊字符的深层列（如Kafka.blob改为kafka_blob），可通过以下两种方式实现：

方法1：修改推断Schema后重新读取数据 #

先让Spark自动推断原始Schema，递归修改列名规则后，用新Schema重新读取数据，效率更高。

步骤代码 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, ArrayType

# 初始化Spark会话
spark = SparkSession.builder.appName("NestedColRename").getOrCreate()

# 1. 读取JSON获取自动推断的Schema
df_temp = spark.read.json("your/json/file/path")
original_schema = df_temp.schema

# 2. 定义列名清理规则：替换点为下划线，转小写
def clean_col_name(name):
    return name.replace(".", "_").lower()

# 3. 递归更新Schema结构
def update_schema(schema):
    if isinstance(schema, StructType):
        new_fields = []
        for field in schema.fields:
            updated_data_type = update_schema(field.dataType)
            new_field = StructField(clean_col_name(field.name), updated_data_type, field.nullable)
            new_fields.append(new_field)
        return StructType(new_fields)
    elif isinstance(schema, ArrayType):
        updated_element_type = update_schema(schema.elementType)
        return ArrayType(updated_element_type, schema.containsNull)
    else:
        # 基本数据类型直接返回
        return schema

# 生成修改后的Schema
updated_schema = update_schema(original_schema)

# 4. 用新Schema重新读取JSON
df_clean = spark.read.schema(updated_schema).json("your/json/file/path")

# 验证结果
df_clean.printSchema()

方法2：已读取DataFrame后递归重命名列 #

如果已经读取了DataFrame，不想重复IO操作，可通过递归构造列表达式来重命名深层嵌套列。

步骤代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, struct, array

# 初始化Spark会话
spark = SparkSession.builder.appName("NestedColRename").getOrCreate()

# 读取原始JSON数据
df = spark.read.json("your/json/file/path")

# 列名清理规则
def clean_col_name(name):
    return name.replace(".", "_").lower()

# 递归构造列表达式
def build_rename_exprs(schema, prefix=""):
    exprs = []
    for field in schema.fields:
        new_name = clean_col_name(field.name)
        full_col_path = f"{prefix}.{field.name}" if prefix else field.name
        
        if isinstance(field.dataType, StructType):
            # 处理嵌套Struct：递归生成子字段表达式，再打包为Struct
            struct_expr = struct(*build_rename_exprs(field.dataType, full_col_path)).alias(new_name)
            exprs.append(struct_expr)
        elif isinstance(field.dataType, ArrayType):
            element_type = field.dataType.elementType
            if isinstance(element_type, StructType):
                # 处理Array内的Struct：递归生成元素的Struct表达式，再打包为Array
                array_expr = array(struct(*build_rename_exprs(element_type, f"{full_col_path}[0]"))).alias(new_name)
                exprs.append(array_expr)
            else:
                # Array内为基本类型，直接重命名
                exprs.append(col(full_col_path).alias(new_name))
        else:
            # 基本类型列直接重命名
            exprs.append(col(full_col_path).alias(new_name))
    return exprs

# 应用重命名规则
df_clean = df.select(*build_rename_exprs(df.schema))

# 验证结果
df_clean.printSchema()

两种方法都能实现深层列名的批量修改，无需手动硬编码复杂的嵌套Schema，适配任意层级的struct和array组合。

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