使用Spark从Kafka读取Avro数据时解析报错该如何解决？

1 优先排查Schema匹配问题 #

这是该类报错最常见的根因：

• 写入数据时使用to_avro(struct(*))会根据源DataFrame的Schema自动生成Avro Schema，需确认你读取时使用的user.avsc和写入时的Avro Schema完全一致，包括字段名大小写、字段类型映射、nullable属性：例如源DataFrame某字段为StringType(nullable=True)，对应的Avro Schema类型应为["null", "string"]而非单纯的"string"，否则会触发解析失败。
• 可直接导出写入时的源DataFrame Schema做对比：执行print(源写入DataFrame.schema.simpleString())，和user.avsc的字段定义逐一核对。

2 正确配置from_avro的容错模式 #

你之前配置的permissive模式作用于CSV读取或Kafka数据源，from_avro的解析模式需要单独在函数参数中配置才会生效，正确写法如下：

from pyspark.sql.avro.functions import from_avro

jsonFormatSchema = open("examples/src/main/resources/user.avsc", "r").read()

df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
.option("subscribe", "topic1") \
.load()

# permissive模式传入from_avro的options参数
output = df.select(
    from_avro(
        data = "value",
        jsonFormatSchema = jsonFormatSchema,
        options = {"mode": "PERMISSIVE"}
    ).alias("user")
)

配置后解析失败的字段会返回null而非直接抛异常，你可以后续过滤user is null的记录单独排查坏数据。

3 排查Avro数据前缀问题 #

如果写入Kafka时使用了Confluent Schema Registry的序列化组件，生成的Avro二进制数据会带5字节的Schema ID前缀，Spark原生from_avro默认解析纯Avro二进制数据，带前缀的内容会直接解析失败，可手动截去前缀后再解析：

import pyspark.sql.functions as F

# 截去前5字节前缀
df = df.withColumn("trimmed_value", F.expr("substring(value, 6)"))
output = df.select(
    from_avro("trimmed_value", jsonFormatSchema, options={"mode": "PERMISSIVE"}).alias("user")
)

4 离线验证单条数据 #

可先导出少量Kafka原始二进制数据，用本地Avro库直接解析，确认数据本身没有损坏、和Schema匹配：

# 先导出原始二进制数据
df.select("value").limit(10).write.format("parquet").save("/tmp/kafka_raw")

用avro库读取单条二进制内容校验，如果本地解析也失败，说明写入环节的Avro数据本身存在异常。

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