# Kafka to ElasticSearch
**URL:** https://discuss.elastic.co/t/kafka-to-elasticsearch/21012
**Category:** Elasticsearch
**Created:** [December 1, 2014, 6:25pm UTC](https://discuss.elastic.co/t/kafka-to-elasticsearch/21012 "2014-12-01T18:25:11Z")
**Posts on this page:** 6
**Page:** 1
### Author:  [@ES\_MacTalk](https://discuss.elastic.co/u/ES_MacTalk)
#### Post date: [December 1, 2014, 6:25pm UTC](https://discuss.elastic.co/t/kafka-to-elasticsearch/21012/1 "2014-12-01T18:25:11Z")
Hi,
We are trying to move data from a Kafka Topic to ElasticSearch.We are getting data in JSON format in the Kafka Topic.We are planning to move this data to Elastic Search and then finally visualize using kibana. We are right now using Flume Elastic search sink using the default serializer for the same.But we are not able to visualize the data in Kibana..
In the elastic search ,the
{
"\_index": "handleindex6-2014-12-01",
"\_type": "bar\_type",
"\_id": "AUoGQAHfOA15NKJPcqsy",
"\_score": 1,
"\_source": {
**"@message": "{"handle":"speed","mac":"WIN1223","owner":"Sam","schema ID":"12B007","data":"19.66","timestamp":1366150981}",**
"@timestamp": "2014-12-01T14:27:45.509Z",
"@fields": {
"topic": "mtalk",
"timestamp": "1417444065509"
}
We are seeing that the data is getting embedded withing the message part of the json saved in elastic search.Is this the right approach Kafka--\>Flume--\>ES--\>Kibana?
---
### Author:  [@Mungeol\_Heo](https://discuss.elastic.co/u/Mungeol_Heo)
#### Post date: [December 3, 2014, 12:08am UTC](https://discuss.elastic.co/t/kafka-to-elasticsearch/21012/2 "2014-12-03T00:08:24Z")
There is another approach which is listed below.
kafka -\> logstash consumer (which is logstash-kafka) -\> es
you can check logstash-kafka at "[GitHub - joekiller/logstash-kafka: Kafka plugin for Logstash](https://github.com/joekiller/logstash-kafka)"
And, I tried the approach which you mentioned above.
However, for me, es sink of flume is kind of unstable.
Anyway, you can try it.
Hope it helps.
On Tue, Dec 2, 2014 at 3:25 AM, ES\_MacTalk [baluvignesh@gmail.com](mailto:baluvignesh@gmail.com) wrote:
> Hi,
> We are trying to move data from a Kafka Topic to Elasticsearch.We are
> getting data in JSON format in the Kafka Topic.We are planning to move this
> data to Elastic Search and then finally visualize using kibana. We are right
> now using Flume Elastic search sink using the default serializer for the
> same.But we are not able to visualize the data in Kibana..
>
> In the Elasticsearch ,the
> {
> "\_index": "handleindex6-2014-12-01",
> "\_type": "bar\_type",
> "\_id": "AUoGQAHfOA15NKJPcqsy",
> "\_score": 1,
> "\_source": {
> _"@message":
> "{"handle":"speed","mac":"WIN1223","owner":"Sam","schema
> ID":"12B007","data":"19.66","timestamp":1366150981}",_
> "@timestamp": "2014-12-01T14:27:45.509Z",
> "@fields": {
> "topic": "mtalk",
> "timestamp": "1417444065509"
> }
> We are seeing that the data is getting embedded withing the message part of
> the json saved in Elasticsearch.Is this the right approach
> Kafka--\>Flume--\>ES--\>Kibana?
>
> --
> View this message in context: [http://elasticsearch-users.115913.n3.nabble.com/Kafka-to-ElasticSearch-tp4066982.html](http://elasticsearch-users.115913.n3.nabble.com/Kafka-to-ElasticSearch-tp4066982.html)
> Sent from the Elasticsearch Users mailing list archive at [Nabble.com](http://Nabble.com).
>
> --
> You received this message because you are subscribed to the Google Groups "elasticsearch" group.
> To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
> To view this discussion on the web visit [https://groups.google.com/d/msgid/elasticsearch/1417458311296-4066982.post%40n3.nabble.com](https://groups.google.com/d/msgid/elasticsearch/1417458311296-4066982.post%40n3.nabble.com).
> For more options, visit [https://groups.google.com/d/optout](https://groups.google.com/d/optout).
--
You received this message because you are subscribed to the Google Groups "elasticsearch" group.
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).
To view this discussion on the web visit [https://groups.google.com/d/msgid/elasticsearch/CADQPeWz%2B7YtaL9rSFqsKfs-dvVDte0RMp%2BnFyu1Q%3Do%2BCUVwjqw%40mail.gmail.com](https://groups.google.com/d/msgid/elasticsearch/CADQPeWz%2B7YtaL9rSFqsKfs-dvVDte0RMp%2BnFyu1Q%3Do%2BCUVwjqw%40mail.gmail.com).
For more options, visit [https://groups.google.com/d/optout](https://groups.google.com/d/optout).
---
### Author:  [@sangala\_hadoop](https://discuss.elastic.co/u/sangala_hadoop)
#### Post date: [May 25, 2015, 6:22am UTC](https://discuss.elastic.co/t/kafka-to-elasticsearch/21012/3 "2015-05-25T06:22:45Z")
I am also trying kafka--\> flume --\> elassticserch. but i am not able to do with below configuration.
Please share flume configuration.
flume1.sources = source1
flume1.channels = channel1
flume1.sinks = elasticsearch
flume1.channels.channel1.type = memory
flume1.channels.channel1.capacity = 10000000
flume1.channels.channel1.transactionCapacity = 1000
# For each source, channel, and sink, set standard properties
flume1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
flume1.sources.source1.zookeeperConnect = hosthere:2181
flume1.sources.source1.topic = kafka\_topicname\_here
flume1.sources.source1.batchSize = 5
flume1.sources.source1.batchDurationMillis = 200
flume1.sources.source1.channels = channel1
flume1.sinks.elasticsearch.channel = channel1
flume1.sinks.elasticsearch.type=elasticsearch
flume1.sinks.elasticsearch.batchSize=100
flume1.sinks.elasticsearch.hostNames = elastic\_host\_here:9300
flume1.sinks.elasticsearch.indexName = foo\_index
flume1.sinks.elasticsearch.indexType = bar\_type
flume1.sinks.elasticsearch.ttl = 2d
flume1.sinks.elasticsearch.clusterName = myclustername\_here
flume1.sinks.elasticsearch.serializer = org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer
---
### Author:  [@sangala\_hadoop](https://discuss.elastic.co/u/sangala_hadoop)
#### Post date: [May 25, 2015, 6:55am UTC](https://discuss.elastic.co/t/kafka-to-elasticsearch/21012/4 "2015-05-25T06:55:48Z")
I fixed ...and now working fine
thanks,
Shekhar Reddy.
---
### Author:  [@sangala\_hadoop](https://discuss.elastic.co/u/sangala_hadoop)
#### Post date: [May 26, 2015, 5:48am UTC](https://discuss.elastic.co/t/kafka-to-elasticsearch/21012/5 "2015-05-26T05:48:51Z")
Hi Mac,
I am using org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer in flume config and now kafka topic data coming in body. please share details if you fixed this issue
flume1.sinks.k1.serializer = org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer
{
\_index: foo\_index-2015-05-25
\_type: bar\_type
\_id: AU2L8-LiEl9820ruvxyR
\_version: 1
\_score: 1
\_source: {
**body: {customer: customer1,network: network1, domain: XXX, kpiGroup:kpiGroup1 ,vendor: vendor1,entityType: entityType1}**
timestamp: 1432572122700
topic: topic\_name
}
}
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [July 6, 2017, 12:11am UTC](https://discuss.elastic.co/t/kafka-to-elasticsearch/21012/6 "2017-07-06T00:11:57Z")