使用PySpark Structured Streaming解析Kafka返回全Null问题排查

尝试用以下PySpark代码解析Kafka流数据，执行后输出全为Null：

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *


spark = SparkSession  \
    .builder  \
    .appName("StructuredSocketRead")  \
    .getOrCreate()
spark.sparkContext.setLogLevel('ERROR') 
    
lines = spark  \
    .readStream  \
    .format("kafka")  \
    .option("kafka.bootstrap.servers","***.compute-1.amazonaws.com:9092")  \
    .option("subscribe","kafka-new-topic")  \
    .load()

# Transform to Output DataFrame
schema = StructType([
        StructField("card_id", LongType()),
        StructField("member_id", LongType()),
        StructField("amount", LongType()),
        StructField("pos_id", LongType()),
        StructField("postcode", LongType()),
        StructField("transaction_dt", StringType()),
    ])

value_df = df.select(from_json(col("value").cast("string"),schema).alias("value"))
exploded_df = value_df.select("value.*")

query = exploded_df  \
    .writeStream  \
    .outputMode("append")  \
    .format("console")  \
    .start()
    
query.awaitTermination()

改用以下代码直接输出value字段内容：

kafkaDF = lines.selectExpr("cast(key as string)","cast(value as string)")

query = kafkaDF   \
    .writeStream  \
    .outputMode("append")  \
    .format("console")  \
    .start()
    
query.awaitTermination()

得到输出：

Batch: 1
-------------------------------------------
+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|value                                                                                                                                                                     |
+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|"{\"card_id\":348702330256514,\"member_id\":37495066290,\"amount\":4380912,\"postcode\":96774,\"pos_id\":248063406800722,\"transaction_dt\":\"01-03-2018 08:24:29\"}\n"   |
|"{\"card_id\":348702330256514,\"member_id\":37495066290,\"amount\":6703385,\"postcode\":84758,\"pos_id\":786562777140812,\"transaction_dt\":\"02-06-2018 04:15:03\"}\n"   |
|"{\"card_id\":348702330256514,\"member_id\":37495066290,\"amount\":7454328,\"postcode\":93645,\"pos_id\":466952571393508,\"transaction_dt\":\"12-02-2018 09:56:42\"}\n"   |

已尝试调整Schema数据类型，问题仍未解决，请问原因是什么？如何解决？

从value字段的输出可以看出，数据是双重转义的JSON字符串：

1. 外层有双引号包裹（"{...}"）
2. JSON内部的双引号被转义成\"
3. 末尾还有换行符\n

这种情况下，直接用from_json解析会因为格式不匹配返回全Null，因为from_json期望标准JSON格式，而非带外层引号和转义的字符串。

另外原代码存在变量错误：使用了未定义的df，实际应该是lines，但该错误会导致代码运行失败，推测你实际运行时已修正该问题，核心矛盾仍是JSON格式的双重转义。

需要先对value字段做字符串清洗，去除外层引号、转义字符和换行符，再用from_json解析。修正后的完整代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *


spark = SparkSession  \
    .builder  \
    .appName("StructuredSocketRead")  \
    .getOrCreate()
spark.sparkContext.setLogLevel('ERROR') 
    
lines = spark  \
    .readStream  \
    .format("kafka")  \
    .option("kafka.bootstrap.servers","***.compute-1.amazonaws.com:9092")  \
    .option("subscribe","kafka-new-topic")  \
    .load()

# 定义Schema
schema = StructType([
        StructField("card_id", LongType()),
        StructField("member_id", LongType()),
        StructField("amount", LongType()),
        StructField("pos_id", LongType()),
        StructField("postcode", LongType()),
        StructField("transaction_dt", StringType()),
    ])

# 清洗value字段：去除首尾引号、替换转义双引号、去除末尾换行
cleaned_value = lines.select(
    regexp_replace(
        regexp_replace(col("value").cast("string"), '^"|"$', ''),
        '\\"', '"'
    ).alias("cleaned_value")
).withColumn("cleaned_value", regexp_replace(col("cleaned_value"), '\n$', ''))

# 解析清洗后的JSON字符串
parsed_df = cleaned_value.select(from_json(col("cleaned_value"), schema).alias("data")).select("data.*")

# 输出到控制台
query = parsed_df  \
    .writeStream  \
    .outputMode("append")  \
    .format("console")  \
    .start()
    
query.awaitTermination()

如果数据的双重转义是因为Kafka生产者端重复序列化导致的，建议从源头解决：确保生产者只做一次JSON序列化，避免生成格式异常的字符串，这样消费端无需额外清洗即可直接解析。

内容的提问来源于stack exchange，提问作者rind-tran