Spark如何处理含点（.）的键？嵌套JSON扁平化报错解决

问题根源 #

PySpark默认会把JSON里带点的键（比如abc.xyz）解析成嵌套Struct结构（误以为abc是父字段、xyz是子字段），但你的JSON中abc.xyz是单个完整字段名，这就导致扁平化时找不到对应的嵌套字段，抛出FIELD_NOT_FOUND错误。

解决方案步骤 #

1. 正确读取JSON，保留带点原始字段名 #

不要直接用spark.read.json()读取（它会自动拆分带点键），先按文本读取每行，再手动指定Schema解析JSON：

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

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

# 定义精确匹配的Schema，带点字段直接用字符串指定
custom_schema = StructType([
    StructField("key1", StructType([
        StructField("key2", ArrayType(
            StructType([
                StructField("key3", StructType([
                    StructField("key4", StructType([
                        StructField("abc.xyz", StringType(), nullable=True)
                    ]))
                ]))
            ])
        ))
    ]))
])

# 按文本读取每行JSON
df_raw = spark.read.text("your_large_file.json")

# 用自定义Schema解析JSON为DataFrame
df = df_raw.selectExpr("from_json(value, '{}') as data".format(custom_schema.json())).select("data.*")

2. 修改自定义扁平化函数，适配带点字段 #

遍历Struct字段时，对带点的字段名用反引号`包裹，避免PySpark误解析为嵌套结构。以下是适配后的函数示例：

from pyspark.sql.types import StructType, ArrayType
from pyspark.sql.functions import col

def flatten_df(df, prefix=""):
    fields = []
    for field in df.schema.fields:
        # 带点字段用反引号包裹
        field_name = f"`{field.name}`" if "." in field.name else field.name
        full_name = f"{prefix}.{field_name}" if prefix else field_name
        
        if isinstance(field.dataType, StructType):
            # 递归处理嵌套Struct
            fields.extend(flatten_df(df.select(field_name), prefix=full_name).columns)
        elif isinstance(field.dataType, ArrayType):
            # 展开数组并递归处理（保留索引可选，按需调整）
            exploded = df.withColumn(f"{full_name}_exploded", col(field_name))
            flattened_array = flatten_df(exploded.select(f"{full_name}_exploded.*"), prefix=full_name)
            fields.extend(flattened_array.columns)
        else:
            # 重命名字段，把点换成下划线避免后续问题
            new_name = full_name.replace(".", "_").replace("`", "")
            fields.append(col(field_name).alias(new_name))
    return df.select(*fields)

# 调用扁平化函数
flattened_df = flatten_df(df)
flattened_df.show()

关键注意事项 #

• 读取阶段必须用自定义Schema强制保留带点字段名，阻止PySpark自动拆分。
• 引用带点字段时必须加反引号，否则会被解析为嵌套结构。
• 最终扁平化后的字段建议替换掉点（比如换成下划线），避免后续SQL操作再次出现解析问题。

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