PySpark如何迭代提取JSON字符串列中id_2对应键的取值

问题背景 #

现有结构化数据集包含id_1、id_2、json_string三列，需要根据每行id_2的取值，从json_string存储的JSON对象中提取对应键的数值，生成value列，样例数据如下：

初始实现尝试通过concat函数动态拼接JSON路径传入get_json_object，运行时触发Column is not iterable报错；硬编码固定JSON路径（如'$.1001'）时代码可正常运行，但实际场景中JSON包含数千个键，无法手动枚举所有路径。

报错根因 #

get_json_object的第二个入参要求是常量字符串或可在静态解析阶段确定值的表达式，不支持直接传入逐行变化的Column类型动态路径，因此Python侧直接拼接列对象生成路径的写法会触发类型校验错误。

可行实现方案 #

方案1：JSON解析为Map后按键取值（推荐，性能最优） #

先通过from_json将JSON字符串统一解析为Map类型列，再直接按行取id_2对应的值即可，仅需一次JSON解析，后续取值无额外开销，适合大数据量场景。

from pyspark.sql import functions as F
from pyspark.sql.types import MapType, StringType, IntegerType

# 定义JSON结构对应的schema，键为字符串类型，值为整数类型，可根据实际值类型调整
json_map_schema = MapType(StringType(), IntegerType())

df_result = df.withColumn(
    # 先把JSON字符串转为Map结构
    "tmp_json_map",
    F.from_json(F.col("json_string"), json_map_schema)
).withColumn(
    "value",
    # 数值类型的id_2需转为字符串，和Map的键类型匹配
    F.col("tmp_json_map")[F.col("id_2").cast(StringType())]
).drop("tmp_json_map") # 删除临时生成的Map列

方案2：通过expr构造动态SQL表达式 #

如果不想提前定义JSON schema，可使用expr将路径拼接逻辑下沉到Spark SQL执行层，绕过Python侧的参数类型校验，写法更轻量化。

from pyspark.sql import functions as F

df_result = df.withColumn(
    "value",
    F.expr("get_json_object(json_string, concat('$.', id_2))")
)

注意：如果JSON键名包含特殊字符（如空格、点号），需要给路径中的键加双引号包裹，路径拼接规则改为concat('$."', id_2, '"')即可。

方案选型建议 #

• 生产环境、数据量较大、JSON结构相对固定的场景优先选择方案1，解析性能更高，也不存在特殊字符转义问题
• 临时探索数据、JSON结构不固定的场景可选择方案2，代码更简洁无需提前定义schema

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