PySpark多主题结构化流读写时输出Null的原因及解决方法

我用PySpark从多个Kafka主题消费结构化流数据，数据的value列是JSON格式字符串，示例如下：

{"message":"abc", "metrics":{"metric1":"abc", "metric2":123, "metric3":"01/01/2022 00:00:00"}}

我通过如下方式从metrics列提取特定字段：

value_schema = StructType([StructField("metrics", StringType(), True)])
topic1_schema = StructType([StructField("metrics1", StringType(), True),
                                         ...
                            StructField("metricsN", StringType(), True)])
topic1_raw = spark \
    .readStream \
     ...
    .load() \
    .selectExpr("CAST(value AS STRING)") \
    .withColumn('value', from_json(col('value'), value_schema)) \
    .withColumn('metrics', from_json(col('value.metrics'), topic1_schema)) \
    .select(col('metrics.*'))

topic1_batch = topic1_raw \
    .writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

topic2_schema = StructType([StructField("metricsA", StringType(), True),
                                         ...
                            StructField("metricsZ", StringType(), True)])
topic2_raw = spark \
    .readStream \
     ...
    .load() \
    .selectExpr("CAST(value AS STRING)") \
    .withColumn('value', from_json(col('value'), value_schema)) \
    .withColumn('metrics', from_json(col('value.metrics'), topic2_schema)) \
    .select(col('metrics.*'))


topic2_batch = topic2_raw \
    .writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

topic1_batch.awaitTermination()
topic2_batch.awaitTermination()

单独运行每个主题的处理代码时能得到预期结果，但同时运行所有主题的流处理代码时，部分字段出现Null值。请问该问题的可能原因是什么？如何解决？

原因分析 #

• Schema对象共享冲突：全局复用的value_schema被多个流共享，PySpark结构化流的内部状态管理可能出现异常，导致JSON解析逻辑出错，部分字段解析失败返回Null。
• 集群资源竞争：多个流同时运行时，CPU、内存资源被抢占，JSON解析过程因资源不足出现部分字段解析失败。
• 跨主题数据混入：如果某个主题收到了不属于自身的消息（比如消息发送错误串流），用对应主题的Schema解析时，不存在的字段会被解析为Null。

解决方案 #

1. 为每个流独立定义Schema #

不要复用外层的value_schema，每个主题的流处理单独定义自己的外层Schema，避免对象共享导致的状态冲突：

# 为Topic1单独定义外层Schema
topic1_value_schema = StructType([StructField("metrics", StringType(), True)])
topic1_raw = spark \
    .readStream \
    ...
    .load() \
    .selectExpr("CAST(value AS STRING)") \
    .withColumn('value', from_json(col('value'), topic1_value_schema)) \
    .withColumn('metrics', from_json(col('value.metrics'), topic1_schema)) \
    .select(col('metrics.*'))

# 为Topic2单独定义外层Schema
topic2_value_schema = StructType([StructField("metrics", StringType(), True)])
topic2_raw = spark \
    .readStream \
    ...
    .load() \
    .selectExpr("CAST(value AS STRING)") \
    .withColumn('value', from_json(col('value'), topic2_value_schema)) \
    .withColumn('metrics', from_json(col('value.metrics'), topic2_schema)) \
    .select(col('metrics.*'))

2. 优化资源分配 #

提交Spark作业时，通过--executor-memory、--executor-cores等参数增加资源配额，同时给每个流设置maxOffsetsPerTrigger控制单批次处理的数据量，缓解资源竞争：

# 示例：设置每个批次最多消费1000条数据
topic1_raw = spark \
    .readStream \
    .option("maxOffsetsPerTrigger", 1000) \
    ...
    .load()

3. 添加数据校验逻辑 #

在解析后增加过滤步骤，剔除不符合Schema的异常数据，避免Null值流入下游：

# 过滤metrics解析为空的数据
topic1_raw = topic1_raw.filter(col('metrics').isNotNull())

# 或者针对关键字段做非空校验
topic1_raw = topic1_raw.filter(col('metrics.metrics1').isNotNull())

4. 隔离流的运行环境 #

为每个主题的流创建独立的SparkSession，彻底隔离每个流的状态和资源：

# 为Topic1创建独立SparkSession
spark1 = SparkSession.builder.appName("Topic1StreamProcessing").getOrCreate()
topic1_raw = spark1.readStream...

# 为Topic2创建独立SparkSession
spark2 = SparkSession.builder.appName("Topic2StreamProcessing").getOrCreate()
topic2_raw = spark2.readStream...

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