Databricks流数据中如何从JSON字符串数组按key获取指定val？

针对你遇到的流式数据JSON数组提取问题，以下是两种可行的解决方法：

方法一：修复自定义UDF #

你之前的UDF报错是因为参数传递方式错误——不能直接将常量（如id=2）与列对象混合传入UDF。可以通过闭包封装常量参数，或者直接在UDF中固定目标条件：

固定条件的UDF（适用于只找key=2的场景） #

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import json

def extract_target_val(json_str):
    try:
        data_list = json.loads(json_str)
        for item in data_list:
            if item.get("key") == 2:
                return item.get("val")
        return None  # 无匹配项时返回空
    except Exception:
        return None

# 注册UDF
get_val_udf = udf(extract_target_val, StringType())

# 应用到流DataFrame（假设你的流表中存储JSON数组的列名为`data`）
stream_df = stream_df.withColumn("target_val", get_val_udf("data"))

通用可配置UDF（支持动态指定目标id、字段名） #

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import json

def create_extract_udf(target_id, id_field, val_field):
    def extract_func(json_str):
        try:
            data_list = json.loads(json_str)
            for item in data_list:
                if item.get(id_field) == target_id:
                    return item.get(val_field)
            return None
        except Exception:
            return None
    return udf(extract_func, StringType())

# 创建针对key=2、提取val字段的UDF
get_val_udf = create_extract_udf(2, "key", "val")
stream_df = stream_df.withColumn("target_val", get_val_udf("data"))

方法二：使用Spark内置函数（无需UDF） #

利用from_json将JSON数组转为Array[Map[String, String]]（兼容元素结构不一致的情况），再结合filter或explode提取目标字段：

方式1：用filter直接过滤数组 #

from pyspark.sql.functions import from_json, col, filter

# 定义schema：数组元素为任意键值对的Map
array_map_schema = "array<map<string, string>>"

# 将JSON字符串列转为数组类型
stream_df = stream_df.withColumn("data_array", from_json(col("data"), array_map_schema))

# 过滤出key=2的元素（注意：Map中值为字符串，需转为int与数字2比较）
stream_df = stream_df.withColumn("matched_items", filter(col("data_array"), lambda x: int(x["key"]) == 2))

# 提取第一个匹配项的val字段
stream_df = stream_df.withColumn("target_val", col("matched_items")[0]["val"])

方式2：用explode炸开数组后过滤聚合 #

如果需要保留原表其他字段，可通过分组聚合提取目标值：

from pyspark.sql.functions import from_json, explode, col, first

stream_df = stream_df.withColumn("data_array", from_json(col("data"), "array<map<string, string>>")) \
    .withColumn("item", explode(col("data_array"))) \
    .filter(int(col("item.key")) == 2) \
    .groupBy("原表需要保留的列名1", "原表需要保留的列名2") \
    .agg(first(col("item.val")).alias("target_val"))

关于流式数据测试的说明 #

你之前遇到的"Queries with streaming sources must be executed with writeStream.start()"报错，是因为流式DataFrame的转换操作必须触发流执行。在预处理阶段，无需写入持久化存储，可通过以下方式测试：

• 在Databricks中直接使用display(stream_df)查看实时结果
• 临时启动控制台输出流：

stream_df.writeStream.format("console") \
    .outputMode("append") \
    .start() \
    .awaitTermination(30)  # 运行30秒后自动停止

对之前尝试方法的补充说明 #

1. 索引访问$.data[1].val：依赖数组顺序，可靠性差，不推荐。
2. JSONPath过滤表达式：Databricks的get_json_object不支持[?(@.xxx==yyy)]这类过滤语法，因此无法生效。
3. 动态创建Struct：流式场景下必须通过writeStream触发执行，不能像批处理直接解析，可通过上述临时测试方式验证结果。
4. 固定Struct Schema：数组元素结构不一致时会丢失字段或报错，改用Map类型可兼容任意结构的元素。

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