我正在尝试使用kafka队列中的序列化avro事件。kafka队列是使用一个简单的java生产者填充的。为清楚起见,我将分享以下三个部分:
avro架构文件
{"namespace": "example.avro",
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type": "string"},
{"name": "favorite_number", "type": ["int", "null"]},
{"name": "favorite_color", "type": ["string", "null"]}
]
}
java生产者代码片段(user.class是使用avro工具生成的)
User user1 = new User();
user1.setName("Alyssa");
user1.setFavoriteNumber(256);
user1.setFavoriteColor("blue");
String topic = "MemoryTest";
// Properties set in 'props'
KafkaProducer<Message, byte[]> producer = new KafkaProducer<Message, byte[]>(props);
ByteArrayOutputStream out = new ByteArrayOutputStream();
DatumWriter<User> writer = new SpecificDatumWriter<User>(User.class);
Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
writer.write(user1, encoder);
encoder.flush();
out.close();
byte[] serializedBytes = out.toByteArray();
producer.send(new ProducerRecord<Message, byte[]>(topic, serializedBytes));
日志存储配置文件
input {
kafka {
zk_connect => "localhost:2181"
topic_id => "MemoryTest"
type => "standard_event"
group_id => "butiline_dash_prod"
reset_beginning => true
auto_offset_reset => smallest
codec => {
avro => {
schema_uri => "/opt/ELK/logstash-1.5.4/bin/user.avsc"
}
}
}
}
output {
stdout {
codec => rubydebug
}
}
问题
管道在logstash级别发生故障。当一个新事件被推送到kafka中时,我在logstash控制台上看到以下内容:
Alyssa�blue {:exception=>#<NoMethodError: undefined method `decode' for ["avro", {"schema_uri"=>"/opt/ELK/logstash-1.5.4/bin/user.avsc"}]:Array>, :backtrace=>["/opt/ELK/logstash-1.5.4/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-1.0.0/lib/logstash/inputs/kafka.rb:169:in `queue_event'", "/opt/ELK/logstash-1.5.4/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-1.0.0/lib/logstash/inputs/kafka.rb:139:in `run'", "/opt/ELK/logstash-1.5.4/vendor/bundle/jruby/1.9/gems/logstash-core-1.5.4-java/lib/logstash/pipeline.rb:177:in `inputworker'", "/opt/ELK/logstash-1.5.4/vendor/bundle/jruby/1.9/gems/logstash-core-1.5.4-java/lib/logstash/pipeline.rb:171:in `start_input'"], :level=>:error}
1条答案
按热度按时间zynd9foi1#
最后找出了错误。而不是这个(如logstash网站所建议的-https://www.elastic.co/guide/en/logstash/current/plugins-codecs-avro.html)
正确的语法是(如插件文档中所建议的)https://github.com/logstash-plugins/logstash-codec-avro/blob/master/developer.md):
我猜语法变了。