Kafka with avro records

Question

I have the following: Source - kafka topic (trans) Channel - memory Sink - Hdfs (avro_event)

The data in the kafka topic trans has been written using a c# producer and has thousands of avro records. When I run my flume consumer, it starts to sink the data to the hdfs. The problem is that the data is in the format: schema data schema data

instead of:

schema data data

I am guessing this is because flume is expecting a record type with {header} {body} whereas the data coming from kafka will just be {body} I know there is a way you can wrap the avro data been written to the topic in an avroFlumeEvent but then it seems that it is no longer a true avro record and maybe a spark consumer or storm will prefer the data in true avro down the line. Is there a way to process this topic so the data is written without multiple schemas everytime flume rolls the data to hdfs?

Colman Colman · Accepted Answer · 2015-03-05T03:53:29

We actually got this working in the end. We were using the microsoft .NET avro library instead of the apache avro library in the C# producer. This meant the avro record was serialized correctly. I also needed to change the flume sink to use "org.apache.flume.sink.hdfs.AvroEventSerializer$Builder" as the sink serializer instead of "avro_event". I also needed to include a flume interceptor connected to the kafka source which pushes the variable "flume.avro.schema.url" into the flume header to be later used by the hdfs sink serializer.

I had a look at camus but it seemed overkill for what we were trying to implement, a basic flume channel connected to a kafka topic which sinks the avro data to the hdfs.

I just ripped the interceptor bit from my java app that builds the flume config in the hopes that it might help others who encounter this problem:

                _flumeFileConfigProperties.put(_agentId+".sources." + _sourceId +".interceptors",_interceptorId);           
                _flumeFileConfigProperties.put(_agentId+".sources." + _sourceId + ".interceptors." + _interceptorId + ".type","static");
                _flumeFileConfigProperties.put(_agentId+".sources." + _sourceId + ".interceptors." + _interceptorId + ".key","flume.avro.schema.url");
                _flumeFileConfigProperties.put(_agentId+".sources." + _sourceId + ".interceptors." + _interceptorId + ".value",_avroProdSchemaLocation +_databaseName + "/" + _topic + "/record/" + _schemaVersion + "/" + _topicName + ".avsc");

Kafka with avro records

2 Answers