Spark批量处理Kafka Avro数据出现Malformed data长度为负错误如何解决

该报错的核心原因是你读取的Kafka Avro消息并非纯Avro二进制数据，前缀包含额外的控制字节：

• 绝大多数生产环境的Kafka Avro消息都基于Confluent Schema Registry规范序列化，这类消息的value字段前固定有5字节的前缀：1字节魔术位（固定值为0）+ 4字节Schema ID。你直接将完整value字段传入from_avro解析时，解析器会将前缀识别为Avro数据的一部分，触发数据格式错误、长度为负的异常。
• 你使用的Schema本身没有问题，错误来源于数据前缀未被正确处理。

根据你的场景选择以下两种方案即可：

方案1：对接Confluent Schema Registry（推荐） #

如果你的环境部署了Schema Registry，直接配置from_avro的Confluent兼容参数，会自动处理5字节前缀，无需手动截取数据：

Dataset<Row> load = sparkSession
        .read()
        .format("kafka")
        .option("kafka.bootstrap.servers", (String) kafkaConfiguration.consumerProperties().get("bootstrap.servers"))
        .option("subscribe", kafkaConfiguration.topicsAsCSV(","))
        .load();

var schema = new String(Files.readAllBytes(Paths.get("schema.avsc")));
// 配置Avro解析参数
Map<String, String> avroOptions = new HashMap<>();
avroOptions.put("avroSchema", schema);
avroOptions.put("confluentAvroReaderSchemaRegistryUrl", "你的Schema Registry服务地址，如http://127.0.0.1:8081");

load.select(from_avro(col("value"), avroOptions).alias("data"))
        .write().json("/tmp/spark/json");

方案2：手动截取数据跳过前缀 #

如果不想对接Schema Registry，直接截取value字段第5位之后的二进制数据再解析即可：

Dataset<Row> load = sparkSession
        .read()
        .format("kafka")
        .option("kafka.bootstrap.servers", (String) kafkaConfiguration.consumerProperties().get("bootstrap.servers"))
        .option("subscribe", kafkaConfiguration.topicsAsCSV(","))
        .load();

var schema = new String(Files.readAllBytes(Paths.get("schema.avsc")));
// substring从第5位开始截取，跳过前5字节前缀
load.select(from_avro(expr("substring(value, 5)"), schema).alias("data"))
        .write().json("/tmp/spark/json");

• 特殊情况说明：如果确认你的消息没有遵循Confluent序列化规范，仍出现该报错，则需要检查生产者的序列化逻辑是否和Schema完全匹配，是否存在字段顺序、类型不一致的问题，或者Kafka集群存储的消息本身存在损坏。

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