无法读取卡夫卡 - Avro Schema消息

任何解决此问题的方法？我无法阅读KAFKA-AVRO Schema消息。 Iam试图从logstash发送消息到KAFKA到HDFS。无法读取卡夫卡 - Avro Schema消息

以下是高科技栈：

Logstash 2.3 - 目前生产的版本
汇合3.0。
插件： a。 Logstash-kafka-Output插件 b。 Logstash编解码器，Avro公司。
饲养员：3.4.6
KAFKA：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"} 
     ] 
}

执行以下命令：

个

开始动物园管理员在其自身终端

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

2启动卡夫卡在其自身终端

在自己的终端

3启动模式注册表

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

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

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

5键入您希望上面的命令前跑后，发送到卡夫卡的日志信息：“Hello World”的

6从消费主题卡夫卡

./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!

请让我知道如何解决这个问题

临屋nks， Upendra

来源

2016-06-08 upendra

你是怎么写/发布到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");

来源

2016-10-20 22:05:53 sdeva

无法读取卡夫卡 - Avro Schema消息

回答

相关问题