如何将Kafka Registry获取的JSON Schema应用于Databricks Kafka流数据摄入

错误原因分析 #

• 第一次报错AnalysisException: Cannot parse the schema in JSON format：你传入的是从Schema Registry拿到的原生Schema字符串（通常为Avro Schema的JSON表示或标准JSON Schema），Spark的from_json函数默认仅支持Spark原生的StructType JSON格式或DDL格式Schema，无法直接识别上述原生Schema格式。
• 第二次报错Py4JException: Method from_json(...) does not exist：你通过json.loads将Schema转换为了Python字典，from_json的第二个参数仅支持StructType对象或Schema字符串，不接受Python字典类型的入参，参数类型不匹配导致方法找不到。
• 额外常见误区：90%以上配合Confluent Schema Registry使用的Kafka场景，消息为Avro二进制序列化，你直接将value字段cast为string会导致二进制数据乱码，本身就是错误操作。

正确实现方案 #

场景1：Kafka消息为Avro序列化（主流场景） #

直接使用Spark内置的from_avro函数解析二进制Avro数据，无需转字符串：

from pyspark.sql.avro.functions import from_avro
from avro.schema import parse

# 你从Kafka Registry获取的原生Avro Schema字符串，即之前可正常print的schema变量
raw_avro_schema = schema
# 解析为Avro Schema对象后转换为符合要求的JSON字符串
parsed_avro_schema = parse(raw_avro_schema).__str__()

# 直接读取二进制value字段用from_avro解析
df = spark.readStream \
    .format("kafka") \
    .options(**options) \
    .load() \
    .select(from_avro(col("value"), parsed_avro_schema).alias("data"))

display(df)

场景2：Kafka消息为JSON序列化，Schema Registry仅存储JSON Schema #

先将标准JSON Schema转换为Spark兼容的StructType对象，再传入from_json：

from pyspark.sql.types import StructType
import json

# 你从Kafka Registry获取的原生JSON Schema字符串
raw_json_schema = schema
# 转换为Spark兼容的StructType
spark_schema = spark.read.json(sc.parallelize([raw_json_schema])).schema

# 解析JSON格式的消息
df = spark.readStream \
    .format("kafka") \
    .options(**options) \
    .load() \
    .select(col("value").cast("string")) \
    .select(from_json(col("value"), spark_schema).alias("data"))

display(df)

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