Confluent Schema Registry默认bytes schema（ID:2）问题及Spark解析报错求助

为什么Schema Registry中存在ID为2的bytes类型Schema？ #

Confluent Schema Registry默认预注册了几个基础数据类型的Schema，其中bytes类型的Schema通常分配的ID就是2。这些内置Schema是Registry的核心组成部分，用于处理无需自定义Schema的原始字节数据场景，无需用户手动上传。

Spark消费时自动选用ID2的bytes Schema的原因及解决方法 #

报错Found bytes, expecting test说明Spark在解析消息时，未识别到匹配指定测试Schema的Avro数据，最终 fallback到了内置的bytes Schema。常见原因及修复方式如下：

1. MQTT-Kafka源连接器配置错误，未正确使用Avro序列化 #

检查连接器配置，确保以下参数设置正确：

• value.converter必须设为io.confluent.connect.avro.AvroConverter
• 必须配置value.converter.schema.registry.url指向你的Schema Registry地址
• 若需自动注册Schema，需设置value.converter.auto.register.schemas=true（或提前手动注册测试Schema）

如果连接器配置错误，会直接将原始MQTT消息以bytes形式写入Kafka，而非带Schema ID的Confluent Avro格式，导致Spark无法识别正确的Schema。

2. Spark消费端配置缺失或错误 #

确保Spark的Kafka消费者配置正确关联Schema Registry：

• 设置spark.sql.streaming.kafka.consumer.value.deserializer为io.confluent.kafka.serializers.KafkaAvroDeserializer
• 配置schema.registry.url参数指向你的Registry地址
• 若要强制使用指定的测试Schema，需在Spark的Avro读取逻辑中明确指定，示例代码：

  import org.apache.spark.sql.avro.functions._
  
  val df = spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "your-kafka-host:port")
    .option("subscribe", "your-topic")
    .load()
    .select(from_avro(col("value"), "com.example.TestSchema", Map("schema.registry.url" -> "your-registry-url")).as("data"))
  

  注：com.example.TestSchema为你测试Schema的全名，也可直接传入Schema的JSON字符串。

3. 消息格式不符合Confluent Avro规范 #

Confluent的Avro消息格式开头是1个固定为0的magic字节 + 4字节的Schema ID，如果MQTT连接器写入的消息不符合该格式，Spark的Avro反序列化器会无法解析Schema ID，进而默认使用bytes类型Schema。可通过查看Kafka消息的原始内容验证这一点。

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