如何用PySpark反序列化读取Kafka消息？读取返回NULL问题求助

问题核心 #

Kafka中存储的是带转义的嵌套JSON字符串（如"{\"sectionName\": \"World news\", ...}"），直接用from_json解析会返回全NULL，但硬编码相同格式的字符串却能正常解析——本质是Kafka传递的消息是双重转义的JSON字符串，PySpark无法直接识别为合法JSON结构。

解决方案 #

方案1：在PySpark中清洗转义字符串后解析

先将Kafka的value字段转为字符串，去除外层引号并还原内部转义的引号，再用from_json解析：

from pyspark.sql.functions import from_json, col, regexp_replace
from pyspark.sql.types import StructType, StringType

# 定义目标Schema
schema = StructType() \
    .add("sectionName", StringType()) \
    .add("webTitle", StringType()) \
    .add("webPublicationDate", StringType()) \
    .add("text", StringType())

# 读取Kafka消息并处理
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-host:port") \
    .option("subscribe", "your-topic") \
    .load() \
    .select(col("value").cast(StringType()).alias("raw_value")) \
    # 1. 去除首尾的双引号 2. 将内部的\"替换为"
    .withColumn("cleaned_json", regexp_replace(regexp_replace(col("raw_value"), "^\"|\"$", ""), "\\\\\"", "\"")) \
    .select(from_json(col("cleaned_json"), schema).alias("data")) \
    .select("data.*")

# 控制台输出验证（流处理场景）
df.writeStream \
    .outputMode("append") \
    .format("console") \
    .start() \
    .awaitTermination()

方案2：修复Kafka生产者的序列化逻辑

如果是生产者发送消息时做了双重序列化（比如连续两次调用json.dumps），直接修改生产者代码，仅对原始JSON对象做一次序列化：

from kafka import KafkaProducer
import json

producer = KafkaProducer(bootstrap_servers=['your-host:port'],
                         value_serializer=lambda x: json.dumps(x).encode('utf-8'))

# 原始数据对象
data = {
    "sectionName": "World news",
    "webTitle": "Russia-Ukraine war live: Russian forces suffering heavy losses but neither side making progress, says UK",
    "webPublicationDate": "2023-11-18T10:24:06Z",
    "text": "test"
}

# 直接发送序列化后的JSON字符串
producer.send('your-topic', value=data)
producer.flush()

验证步骤

先确认读取到的原始消息格式，排查问题根源：

# 批处理方式读取Kafka消息，查看原始字符串
df_raw = spark.read \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-host:port") \
    .option("subscribe", "your-topic") \
    .load() \
    .select(col("value").cast(StringType()).alias("raw_value"))

df_raw.show(truncate=False)

如果输出的raw_value是带外层引号和内部转义的字符串，用方案1处理；如果是正常JSON格式，再检查Schema是否匹配。

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