PySpark处理Kafka流数据：JSON列展开为多列返回NULL求助

可能的原因及解决步骤 #

1. 检查JSON键名与Schema的大小写一致性 #

Spark的from_json对JSON键名是大小写敏感的。如果Kafka中实际的JSON数据键是小写（比如{"a":"xxx", "b":"yyy"}），但你定义的Schema用了大写的A、B，就会导致解析失败返回NULL。

• 先执行以下代码查看原始JSON内容：

  json_df.show(truncate=False)
  

  确认JSON的键名、结构是否和你定义的json_schema完全匹配，调整Schema的StructField名称与JSON键名一致。

2. 验证JSON格式合法性 #

如果JSON本身格式不规范（比如用单引号代替双引号、字段名未加引号、存在多余逗号等），from_json会解析失败返回NULL。

• 用get_json_object测试单个字段的解析：

  from pyspark.sql.functions import get_json_object
  
  json_df.select(get_json_object(col("value"), "$.A").alias("A")).show()
  

  如果该查询返回NULL，说明JSON格式有问题，需要检查Kafka Producer发送的JSON是否符合标准格式（必须用双引号包裹键名和字符串值）。

3. 添加JSON解析选项 #

如果你的JSON存在非标准格式（比如允许单引号、无引号字段名），可以在from_json中指定解析选项：

from pyspark.sql.functions import from_json

json_expanded_df = json_df.withColumn(
    "value", 
    from_json(
        json_df["value"], 
        json_schema,
        options={"allowUnquotedFieldNames": "true", "allowSingleQuotes": "true"}
    )
).select("value.*")

4. 确认数据类型完全匹配 #

即使修改Schema类型后还是NULL，要确认JSON中的值类型和Schema定义一致：

• 如果JSON中字段值是字符串形式的数字（比如"A":"123"），Schema必须用StringType，如果用IntegerType会解析失败；
• 如果JSON中是数字类型（比如"A":123），Schema用IntegerType才能正确解析。

修正后的示例代码 #

假设你的JSON格式是{"a":"test", "b":"demo"}，调整Schema并验证：

from pyspark.sql.types import StructType, StructField, StringType
from pyspark.sql.functions import from_json

# 调整Schema键名为小写，匹配JSON实际结构
json_schema = StructType(
    [
        StructField('a', StringType(), True),
        StructField('b', StringType(), True)
    ]
) 

json_df = df.selectExpr("cast(value as string) as value")
# 先查看原始JSON内容
json_df.show(truncate=False)

json_expanded_df = json_df.withColumn(
    "value", 
    from_json(json_df["value"], json_schema)
).select("value.*") 

json_expanded_df.show()

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