如何在Spark Streaming中解析Kafka动态JSON格式消息

方法1：基于样本数据自动推断Schema（Spark 2.3+支持） #

如果能获取到一条具有代表性的JSON样本，可通过schema_of_json自动生成Schema，再传入from_json完成解析：

1. 准备样本JSON字符串（可从Kafka主题提取或手动构造）
2. 生成推断Schema并解析流数据：

from pyspark.sql.functions import from_json, col, schema_of_json, lit

# 示例样本JSON
sample_json = '{"id":1,"name":"demo","info":{"age":30,"city":"Beijing"}}'
# 生成推断Schema
inferred_schema = schema_of_json(lit(sample_json))
# 解析Kafka消息中的JSON数据
df = df.select(from_json(col("value").cast("string"), inferred_schema).alias("parsed_value"))

方法2：解析为Map类型（完全动态适配） #

若JSON结构无固定规律，可直接将其解析为键值对Map，后续按需提取字段：

from pyspark.sql.functions import from_json, col
from pyspark.sql.types import MapType, StringType, ObjectType

# 解析为<String, String>类型Map，适用于值均为字符串的场景
df = df.select(from_json(col("value").cast("string"), MapType(StringType(), StringType())).alias("parsed_value"))

# 若值包含多种类型，可使用<ObjectType>（需注意后续处理时的类型判断）
df = df.select(from_json(col("value").cast("string"), MapType(StringType(), ObjectType())).alias("parsed_value"))

后续可通过parsed_value["字段名"]提取对应内容，嵌套字段需逐层遍历。

方法3：按需提取单个字段（无需全量解析） #

如果仅需获取JSON中的特定字段，可使用get_json_object直接提取，无需定义Schema：

from pyspark.sql.functions import get_json_object, col

# 提取顶层字段id
df = df.select(get_json_object(col("value").cast("string"), "$.id").alias("id"))
# 提取嵌套字段info.city
df = df.select(get_json_object(col("value").cast("string"), "$.info.city").alias("city"))

注意事项 #

• 自动推断Schema依赖样本数据的完整性，若后续消息出现样本中未包含的字段，解析时会被忽略。
• Map类型解析能保留所有字段，但处理嵌套结构时需要手动遍历，灵活性高但易用性弱于强类型Schema。

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