PySpark DataFrame提取字典列表中isCat字段生成新列

getItem()只能按数组索引取固定位置的元素，无法根据字典的Name字段做匹配筛选，所以得通过解析JSON数组→过滤目标元素→提取值的流程实现需求，具体步骤如下：

1. 定义JSON数组的Schema #

先明确CustomFields解析后的结构：它是由包含Name和Value字段的字典组成的数组，定义对应的PySpark Schema：

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

custom_fields_schema = ArrayType(
    StructType([
        StructField("Name", StringType()),
        StructField("Value", StringType())
    ])
)

2. 解析JSON字符串为可操作的数组类型 #

用from_json()函数把CustomFields列的JSON字符串转换成PySpark能识别的数组结构：

df = df.withColumn("custom_fields_array", from_json(df.CustomFields, custom_fields_schema))

3. 过滤并提取目标字段值 #

使用filter()函数筛选数组中Name等于"isCat"的元素，再用element_at()取过滤后的第一个匹配元素的Value，最后用coalesce()处理空值场景（数组为空或无匹配元素时返回"no"）：

from pyspark.sql.functions import filter, element_at, coalesce, col

df = df.withColumn(
    "isCat",
    coalesce(
        element_at(filter(col("custom_fields_array"), lambda x: x.Name == "isCat"), 1).Value,
        "no"
    )
)

4. （可选）清理中间列 #

如果不需要中间生成的custom_fields_array列，可以直接删除：

df = df.drop("custom_fields_array")

完整示例代码 #

from pyspark.sql import SparkSession
from pyspark.sql.types import ArrayType, StructType, StructField, StringType, IntegerType
from pyspark.sql.functions import from_json, filter, element_at, coalesce, col

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

# 模拟测试数据
data = [
    (1, '[{"Name":"isCat","Value":"yes"},{"Name":"color","Value":"orange"}]'),
    (2, '[{"Name":"color","Value":"black"}]'),
    (3, "[]"),
    (4, '[{"Name":"isCat","Value":"true"}]')
]

df = spark.createDataFrame(data, ["Id", "CustomFields"])

# 定义Schema
custom_fields_schema = ArrayType(
    StructType([
        StructField("Name", StringType()),
        StructField("Value", StringType())
    ])
)

# 解析JSON并生成isCat列
df = df.withColumn("custom_fields_array", from_json(df.CustomFields, custom_fields_schema)) \
       .withColumn(
           "isCat",
           coalesce(
               element_at(filter(col("custom_fields_array"), lambda x: x.Name == "isCat"), 1).Value,
               "no"
           )
       ) \
       .drop("custom_fields_array")

# 查看结果
df.show(truncate=False)

运行后输出结果：

+---+------------------------------------------------+-----+
|Id |CustomFields                                    |isCat|
+---+------------------------------------------------+-----+
|1  |[{"Name":"isCat","Value":"yes"},{"Name":"color","Value":"orange"}]|yes|
|2  |[{"Name":"color","Value":"black"}]              |no   |
|3  |[]                                              |no   |
|4  |[{"Name":"isCat","Value":"true"}]               |true |
+---+------------------------------------------------+-----+

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