# Flume to elastic data transfer avro conversion
**URL:** https://discuss.elastic.co/t/flume-to-elastic-data-transfer-avro-conversion/106110
**Category:** Elasticsearch
**Created:** [November 2, 2017, 2:26am UTC](https://discuss.elastic.co/t/flume-to-elastic-data-transfer-avro-conversion/106110 "2017-11-02T02:26:26Z")
**Posts on this page:** 2
**Page:** 1
### Author:  [@putharekulu](https://discuss.elastic.co/u/putharekulu)
#### Post date: [November 2, 2017, 2:26am UTC](https://discuss.elastic.co/t/flume-to-elastic-data-transfer-avro-conversion/106110/1 "2017-11-02T02:26:26Z")
Hello,
I am trying to use flume to send data to elastic . Flume collects the logs from kafka and sinks it to elastic. log format in kafka is in avro. Data that goes to elastic has to be in json format. can anyone suggest me an existing interceptor/serializer to convert avro to elastic suitable format .
Here is what i have so far.
agent.sources=test\_source
agent.channels=test\_channel
agent.sinks=test\_sync
agent.sources.test\_source.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.test\_source.zookeeperConnect = prod-zk1.internal:2181/kafka
agent.sources.test\_source.topic = test\_elastic
agent.sources.test\_source.groupId = test
agent.sources.test\_source.consumer.timeout.ms = 5000
agent.sources.test\_source.batchDurationMillis = 1000
agent.sources.test\_source.batchSize = 1000
agent.sources.test\_source.auto.commit.enabled = true
agent.sources.test\_source.interceptors = i1
agent.sources.test\_source.interceptors.i1.type = static
agent.sources.test\_source.interceptors.i1.key = flume.avro.schema.url
agent.sources.test\_source.interceptors.i1.value = hdfs://nameservice/user/flafka/avro\_templates/test\_elastic.avsc
agent.sources.test\_source.channels = test\_channel
agent.channels.test\_channel.type = memory
agent.channels.test\_channel.capacity = 100000
agent.channels.test\_channel.transactionCapacity = 1000
agent.sinks.test\_sync.type = elasticsearch
agent.sinks.test\_sync.hostNames = prod-elastic.internal:9300
agent.sinks.test\_sync.indexName = test
agent.sinks.test\_sync.indexType = test
agent.sinks.test\_sync.clusterName = prod
agent.sinks.test\_sync.batchSize = 1000
agent.sinks.test\_sync.serializer = org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer
agent.sinks.test\_sync.channel = test\_channel
Thanks in advance.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [November 30, 2017, 2:26am UTC](https://discuss.elastic.co/t/flume-to-elastic-data-transfer-avro-conversion/106110/2 "2017-11-30T02:26:38Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.