文件数据发送至Kafka后Spark Structured Streaming消费格式异常排查

• 核心问题出在producer.py的发送逻辑：从文件读取的每一行line本身已经是合法的JSON字符串，不需要再调用json.dumps()进行序列化。json.dumps()接收字符串参数时，会把字符串整体作为JSON值进行转义，最终发送到Kafka的内容就会外层多一层引号、内部双引号被转义，和你观察到的异常现象完全匹配。
• 次要问题：consumer.py中定义的Schema把数值类字段（precio_actual/bid/ask等）都声明为StringType，会导致后续数值计算需要额外转换，而且Spark默认解析JSON时不识别NaN值，需要额外配置。

1. 修改producer.py代码 #

把发送行的逻辑修改为直接对原JSON字符串编码即可，不需要json.dumps：

import json
from kafka import KafkaProducer
import settings

def producer():
    kafka_producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
    with open(settings.HISTORICAL_DATA_FOLDER+"historical_data_level_1.json") as historical_data_level_1:
        for line in historical_data_level_1:
            # 去掉json.dumps，直接对原字符串编码，strip用于去除多余换行符
            kafka_producer.send("test", line.strip().encode('utf-8'))
    # 确保所有消息都发送完成再退出
    kafka_producer.flush()

if __name__ == '__main__':
    producer()

2. （可选）优化consumer.py的Schema和JSON解析逻辑 #

调整字段类型，添加NaN支持，确保解析正常：

# 调整Schema，数值类字段使用对应类型
schema = StructType([
    StructField('localSymbol', StringType()),
    StructField('time', StringType()),
    StructField('precio_actual', DoubleType()),
    StructField('bid', DoubleType()),
    StructField('ask', DoubleType()),
    StructField('high', DoubleType()),
    StructField('low', DoubleType()),
    StructField('close', DoubleType()),
])

# 修改from_json调用，添加允许非数字（NaN）的配置
value_df = kafka_df.select(
    from_json(
        col("value").cast("string"), 
        schema,
        options={"allowNonNumericNumbers": "true"}
    ).alias("value")
)

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