PAYATHON 2026

如何在 Pyspark 中展平复杂的嵌套 JSON 结构？

技术老齐

现有的读取方式可以保留。locations[*].address 和 company.locations[*].address 在接口响应中是字符串，里面存的是 JSON 文本。要继续展开这些字段，需要先用 from_json() 将它们转换为 StructType。

展平时要区分两种操作：展开 struct 会增加列，展开 array 通常会增加行。这个响应包含多个相互独立的数组，如果全部展开到同一张表，就会产生大量组合行。因此，展平前需要确定最终每一行代表什么。

为什么 address 会被读成 string？

接口返回的数据相当于：

{
  "locations": [
    {
      "id": 1,
      "address": "{\"formatted_address\":\"Sofia, Bulgaria\",\"geometry\":{\"location\":{\"lat\":42.6977082,\"lng\":23.3218675}}}"
    }
  ]
}

外层的 locations 是 JSON 数组，数组中的元素是对象；address 的值是一段带引号的文本，所以 Spark 将它推断为 StringType。

json.dumps(data) 会把 Python 对象序列化为 JSON，但不会递归解析字符串中的 JSON。更换读取方式也不会自动去掉这一层字符串编码。

spark.read.json() 负责解析数据和推断 schema，不会自动把所有嵌套字段展开成数百列。可以用 printSchema() 查看读取后的原始结构，再与展平结果对照。

先把 elements 转换为职位记录

以下代码假设 data 是 response.json() 返回的 Python 字典，且运行环境支持原代码使用的 RDD API：

import json
from pyspark.sql import functions as F
from pyspark.sql import types as T

raw_df = spark.read.json(
    spark.sparkContext.parallelize([
        json.dumps(data, ensure_ascii=False)
    ])
)

jobs = (
    raw_df
    .select(F.explode("elements").alias("job"))
    .select("job.*")
)

jobs.printSchema()

此时，jobs 中每行代表一个职位。company 仍是结构体，locations、requirements 等字段仍是数组。

page、totalPages 和 totalElements 是响应级元数据，以上代码没有将它们放入职位表。如果需要保留，还要另外选择这些字段。

分别提取职位地点和公司地点

职位地点与公司地点的数量可能不同，可以分别生成明细表，通过 job_id 关联职位：

def extract_locations(jobs_df, path):
    return (
        jobs_df
        .select(
            F.col("id").alias("job_id"),
            F.col("companyId").alias("company_id"),
            F.posexplode_outer(F.col(path)).alias(
                "location_pos", "location"
            )
        )
        .select(
            "job_id",
            "company_id",
            "location_pos",
            "location.*"
        )
        .withColumnRenamed("address", "address_raw")
    )

job_locations = extract_locations(jobs, "locations")
company_locations = extract_locations(jobs, "company.locations")

posexplode_outer() 在展开数组的同时保留下标 location_pos。下标从 0 开始，可用于追踪元素在原始数组中的顺序。

遇到空数组或 null 数组时，posexplode_outer() 会保留一行，地点相关字段为 null。如果明细表只需要实际存在的地点，可以再做一次过滤：

job_locations = job_locations.filter(
    F.col("location_pos").isNotNull()
)

用 from_json() 解析地址字符串

示例中，geometry.bounds 和 geometry.viewport 至少有两种表示方式。

一种使用坐标对象：

{
  "northeast": {"lat": 42.7877752, "lng": 23.4569049},
  "southwest": {"lat": 42.6030891, "lng": 23.1909885}
}

另一种使用方向数值：

{
  "south": 42.6030891,
  "west": 23.1909885,
  "north": 42.7877752,
  "east": 23.4569049
}

只取第一条地址推断 schema，可能会遗漏另一种结构。接口格式已知时，可以显式定义同时包含这两种表示的 schema：

point_schema = "STRUCT<lat: DOUBLE, lng: DOUBLE>"

extent_schema = f"""
STRUCT<
    northeast: {point_schema},
    southwest: {point_schema},
    north: DOUBLE,
    south: DOUBLE,
    east: DOUBLE,
    west: DOUBLE
>
"""

address_schema = f"""
STRUCT<
    address_components: ARRAY<STRUCT<
        long_name: STRING,
        short_name: STRING,
        types: ARRAY<STRING>
    >>,
    formatted_address: STRING,
    geometry: STRUCT<
        bounds: {extent_schema},
        location: {point_schema},
        location_type: STRING,
        viewport: {extent_schema}
    >,
    place_id: STRING,
    types: ARRAY<STRING>
>
"""

job_locations = job_locations.withColumn(
    "address",
    F.from_json(F.col("address_raw"), address_schema)
)

company_locations = company_locations.withColumn(
    "address",
    F.from_json(F.col("address_raw"), address_schema)
)

job_locations.printSchema()

解析后，address 是结构体，可以直接访问内部字段：

location_summary = job_locations.select(
    "job_id",
    "company_id",
    "location_pos",
    F.col("id").alias("location_id"),
    F.col("address.formatted_address").alias("formatted_address"),
    F.col("address.geometry.location.lat").alias("latitude"),
    F.col("address.geometry.location.lng").alias("longitude")
)

display(location_summary)

某条地址缺少 schema 中的字段时，对应值为 null。保留 address_raw 方便排查解析异常，也便于以后调整 schema。

这份 schema 覆盖了示例中的地址结构。如果完整响应还有其他字段，需要继续补充；未列入 schema 的字段不会出现在解析结果中。

递归展开结构体，按需展开数组

下面的函数只展开 StructType，用完整字段路径生成列名，数组保持原样：

def quoted_col(name):
    return F.col("`" + name.replace("`", "``") + "`")


def flatten_structs(df):
    columns = []
    names = []

    def visit(column, data_type, path):
        if isinstance(data_type, T.StructType):
            for field in data_type.fields:
                visit(
                    column.getField(field.name),
                    field.dataType,
                    path + [field.name]
                )
        else:
            name = "_".join(path)
            names.append(name)
            columns.append(column.alias(name))

    for field in df.schema.fields:
        visit(
            quoted_col(field.name),
            field.dataType,
            [field.name]
        )

    # 也防止忽略大小写时出现歧义。
    if len(names) != len({name.casefold() for name in names}):
        raise ValueError("展平后的列名发生冲突，请调整命名规则")

    return df.select(*columns)


flat_locations = flatten_structs(job_locations)
display(flat_locations)

结果中会包含这些列：

address_formatted_address
address_geometry_location_lat
address_geometry_location_lng
address_geometry_bounds_northeast_lat
address_geometry_bounds_north

address_address_components 和 address_types 仍是数组。如果也需要将它们展开成行，可以在每轮展开结构体后，逐个处理数组：

def flatten_all(df):
    while True:
        df = flatten_structs(df)

        array_field = next(
            (
                field
                for field in df.schema.fields
                if isinstance(field.dataType, T.ArrayType)
            ),
            None
        )

        if array_field is None:
            return df

        df = df.withColumn(
            array_field.name,
            F.explode_outer(quoted_col(array_field.name))
        )


# 确认需要所有数组元素的组合后再执行。
fully_flat_locations = flatten_all(job_locations)

这个函数处理 StructType 和 ArrayType，不会自动识别其他字符串中的 JSON，也不处理 MapType。其他以字符串保存的 JSON 字段，仍需先使用各自的 schema 调用 from_json()。

展平前需要注意什么？

先检查展开数组后的行数。假设某个职位有 2 个地点、5 张产品图片和 3 条任职要求，依次展开这三个数组会得到：

2 × 5 × 3 = 30 行

每行代表一个“地点、图片、要求”的组合。响应没有说明这些数组元素存在一一对应关系，不能直接把这些组合当作业务明细。

这类数据通常可以保留一张每个职位一行的主表，再按 job_id 拆出地点、图片、要求等明细表。如果接口约定两个数组按下标对应，可以考虑使用 arrays_zip()，但使用前必须确认这一约定。

还需要检查以下几点：

  • from_json() 的宽松解析可能返回 null，也可能返回部分字段为 null 的结果。排查时要结合原始字符串和预期必填字段，不能只检查整个结构体是否为 null。开发阶段也可以设置 {"mode": "FAILFAST"}，让解析错误在执行时暴露。
  • 示例响应带有分页元数据。读取一次响应只会处理本次返回的 elements。要采集完整数据，还需按接口的分页规则获取其他页。
  • 将整个响应放进一个字符串，适合小规模交互处理，但数据会先进入 Driver 内存，parallelize() 不会改变这一点。数据量较大时，更适合按记录保存为 JSON Lines 文件，再由 Spark 读取。对于跨多行保存的完整 JSON 文档，再考虑 multiLine 读取选项。
  • 自动推断的 schema 受输入数据影响。字段值全为 null、数组始终为空，或字段只在其他页出现时，都需要特别检查。稳定的生产流程应维护明确的 schema。

备注：内容仅供参考。