PySpark中如何重命名StructType数组字段并设置nullable？

1. 定义法语-英文字段映射表 #

先明确所有需要转换的字段对应关系：

from pyspark.sql import functions as F

fr_eng_mappings = {
    "statutDiffusionUnite": "broadcast_status",
    "unitePurgeeUnite": "is_purged",
    "dateCreationUnite": "creation_date",
    "sigleUnite": "symbol",
    "sexeUnite": "sex",
    "periodesUnite": "UnitPeriods",
    "dateFin": "end_date",
    "dateDebut": "start_date",
    "etatAdministratifUnite": "administrative_state"
}

2. 处理嵌套数组的字段翻译 #

你之前用F.lit(None)生成空值的方式无法填充原数据，需要用transform遍历数组，结合struct重命名字段：

先处理内层periodesUnite的结构体翻译 #

# 定义单个period结构体的翻译逻辑
translate_period = F.struct(
    *[F.col(f"`{col}`").alias(fr_eng_mappings[col]) 
      for col in ["dateFin", "dateDebut", "etatAdministratifUnite"]]
)

再处理外层unites数组，生成Units列 #

遍历unites中的每个结构体，重命名字段同时替换内部的periodesUnite为翻译后的数组：

# 定义单个unit结构体的翻译逻辑，包含嵌套period数组处理
translate_unit = F.struct(
    F.col("id"),
    F.col("score"),
    F.col("statutDiffusionUnite").alias("broadcast_status"),
    F.col("unitePurgeeUnite").alias("is_purged"),
    F.col("dateCreationUnite").alias("creation_date"),
    F.col("sigleUnite").alias("symbol"),
    F.col("sexeUnite").alias("sex"),
    # 翻译嵌套的periodesUnite数组
    F.transform("periodesUnite", lambda p: translate_period).alias("UnitPeriods")
)

# 生成最终的Units列
new_df = df.withColumn("Units", F.transform("unites", lambda u: translate_unit))

3. 生成单独的UnitsPeriods列（扁平化嵌套数组） #

如果需要将所有unites.periodesUnite提取为独立列：

new_df = new_df.withColumn(
    "UnitsPeriods",
    F.flatten(F.transform("unites", lambda u: F.transform(u.periodesUnite, lambda p: translate_period)))
)

4. 设置列的nullable属性 #

Spark中列的nullable默认由表达式自动推断（引用原nullable=true的字段时，新列也会继承该属性）。若要强制修改，需手动更新Schema：

from pyspark.sql.types import StructType, StructField

# 获取当前Schema并修改目标列的nullable属性
updated_fields = []
for field in new_df.schema.fields:
    if field.name == "UnitsPeriods":
        # 重新构造字段，设置nullable为需要的布尔值
        updated_field = StructField(field.name, field.dataType, nullable=True)
        updated_fields.append(updated_field)
    else:
        updated_fields.append(field)

# 应用修改后的Schema
new_df = new_df.sql_ctx.createDataFrame(new_df.rdd, StructType(updated_fields))

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