无法读取 Kafka - Avro Schema 消息答案

【问题标题】：Not able to read Kafka - Avro Schema messages无法读取 Kafka - Avro Schema 消息
【发布时间】：2016-10-08 20:20:16
【问题描述】：

有什么办法解决这个问题？？？我无法阅读 KAFKA-AVRO Schema 消息。我正在尝试将消息从 logstash 发送到 KAFKA 到 HDFS。

以下是技术栈：

Logstash 2.3 - 当前生产版本
融合 3.0。
插件：一种。 Logstash-kafka-输出插件湾。 Logstash-编解码器-avro。
动物园管理员：3.4.6
卡夫卡：0.10.0.0

Logstash 配置文件如下所示：

input {
stdin{}
}

filter {
mutate {
remove_field => ["@timestamp","@version"]
  }
}

output {
  kafka {
topic_id => 'logstash_logs14'

codec => avro  { 
schema_uri => "/opt/logstash/bin/schema.avsc"
    }
  }
}

schema.avsc 文件如下所示：

{
    "type":"record",
    "name":"myrecord",
    "fields":[
        {"name":"message","type":"string"},
        {"name":"host","type":"string"}
        ]
}

运行了以下命令：

在自己的终端中启动 Zookeeper

./bin/zookeeper-server-start ./etc/kafka/zookeeper.properties

2 在自己的终端启动Kafka

./bin/kafka-server-start ./etc/kafka/server.properties

3 在自己的终端中启动模式注册表

./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties

4 从logstash目录，运行以下命令

bin/logstash -f ./bin/logstash.conf

5 输入您希望在运行上述命令后发送到 kafka 的日志消息例如：“你好世界”

6 从 Kafka 消费主题

./bin/kafka-avro-console-consumer --zookeeper localhost:2181 --topic logstash_logs14 --from-beginning
_While consuming we get the following error:_

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/root/confluent-3.0.0/share/java/kafka-serde-tools/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/root/confluent-3.0.0/share/java/confluent-common/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/root/confluent-3.0.0/share/java/schema-registry/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
Processed a total of 1 messages
[2016-06-08 18:42:41,627] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$:103)
org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
[2016-06-08 18:42:41,627] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$:103)
org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

请告诉我如何解决这个问题

谢谢，乌彭德拉

【问题讨论】：

标签： logstash apache-kafka avro

【解决方案1】：

也许回答太晚了，但现在面临同样的问题。

Logstrash 在这里使用默认的序列化器， "org.apache.kafka.common.serialization.StringSerializer"

因此，如果您想从事件总线读取 Avro 消息，则必须在 logstash 输出上使用 KafkaAvroSerializers 对其进行序列化 "io.confluent.kafka.serializers.KafkaAvroSerializer"

然后从消费者部分使用匹配的反序列化器。问题是logstash根本不识别IO.CONFLUENT，所以你必须做一些棘手的事情来添加它，作为deps和jars

【讨论】：

解决了 KafkaAvroSerializer 问题在 LS_HOME/vendor/lib/ 中丢弃 jar 在类路径 Vendor/lib 中添加每个 jar 看起来像：avro-1.10.0.jar common-config-5.5.1.jar jni/ jruby .jar kafka-avro-serializer-5.3.0.jar kafka-clients-2.5.0.jar kafka-schema-registry-client-5.3.0.jar kafka-streams-avro-serde-5.5.1.jar ruby/ discuss.elastic.co/t/…

【解决方案2】：

您如何在 Kafka 上写作/发布？您看到 SerializationException 是因为数据不是使用模式注册表（或 KafkaAvroSerializer）写入的，但是在使用模式注册表时，kafka-avro-console-consumer 在内部使用模式注册表（或 KafkaAvroDeserializer），它期望数据是以某种格式（特别是<magic byte><schemaId><data>）。如果您使用 kafka-avro-console-producer 写入 avro 数据，那么您不应该收到此异常，或者您可以在您的生产者属性中为键和值序列化程序设置 KafkaAvroSerializer 并设置 schema-registry-url。

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
      io.confluent.kafka.serializers.KafkaAvroSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
      io.confluent.kafka.serializers.KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");

【讨论】：