如何解析Kafka中的Avro数据？Druid接入Kafka Avro遇乱码、解析报错问题排查

核心报错触发点 #

• 配置逻辑冲突：当前spec同时在dataSchema下配置了avro_stream类型parser，又在ioConfig下配置了avro_ocf类型inputFormat。Druid 0.20.2版本的Kafka索引任务规则为：只要ioConfig中存在inputFormat配置，就完全忽略dataSchema下的parser配置，直接走InputFormat解析流程，两套解析逻辑并行触发会直接抛出Error: undefined。
• 格式不匹配：avro_ocf仅支持解析带Avro容器头（含schema元数据、同步标记的OCF封装格式）的消息。你的Kafka生产者直接发送Avro序列化裸字节，没有OCF封装头，完全不匹配该InputFormat的解析要求。
• 必选字段传值错误：dataSchema下的timestampSpec、dimensionsSpec被设为null，Druid强制要求这两个字段非空，即使开启自动探测也需要传入空对象，null值会直接触发配置校验失败。
• 自定义解析器不生效原因：只要ioConfig中配置了inputFormat，Druid就不会加载原有parser链路的自定义实现，自然不会进入你编写的ExampleByteBufferInputRowParser的parseBatch方法；若移除inputFormat后仍未触发方法执行，需确认自定义扩展是否已放入extensions目录、且在common.runtime.properties的druid.extensions.loadList中添加了对应扩展包名。

修正后的spec关键配置 #

去掉冲突的parser、inputFormat配置，匹配裸Avro字节流的解析逻辑，关键片段如下：

{
  "type": "kafka",
  "spec": {
    "dataSchema": {
      "dataSource": "new-data-source",
      "timestampSpec": {
        "column": "timestamp",
        "format": "auto"
      },
      "dimensionsSpec": {
        "dimensions": ["id", "name"]
      },
      "metricsSpec": [],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "none",
        "rollup": true
      },
      "parser": {
        "type": "avro_stream",
        "avroBytesDecoder": {
          "type": "schema_inline",
          "schema": {
            "namespace": "com.airebroker.data",
            "name": "Test",
            "type": "record",
            "fields": [
              {"name": "id", "type": "int"},
              {"name": "name", "type": "string"},
              {"name": "timestamp", "type": "long"}
            ]
          }
        },
        "parseSpec": {
          "format": "avro",
          "timestampSpec": {"column": "timestamp", "format": "auto"},
          "dimensionsSpec": {"dimensions": ["id", "name"]}
        }
      }
    },
    "ioConfig": {
      "topic": "TestTopic0409",
      "replicas": 1,
      "taskCount": 1,
      "taskDuration": "PT3600S",
      "consumerProperties": {
        "bootstrap.servers": "localhost:9092"
      },
      "pollTimeout": 100,
      "startDelay": "PT5S",
      "period": "PT30S",
      "useEarliestOffset": false,
      "completionTimeout": "PT1800S"
    }
  }
}

乱码问题排查点 #

• 确认Kafka生产者端的ProducerRecord发送逻辑正确：示例代码中直接将testSchema序列化对象作为value传入，无额外包装层，无需调整生产端代码。
• 确认Kafka生产者端序列化Avro时未额外添加编码转换逻辑，默认Avro string类型使用UTF-8编码，Druid侧解析时不需要额外设置binaryAsString参数。
• 若name字段仍出现乱码，可临时在parser配置中添加binaryAsString: true，确认是否为字节解码逻辑不匹配导致的问题。

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