# Kafka Consumer ElasticSearch?

**URL:** <https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765>\
**Category:** Elasticsearch\
**Created:** [May 1, 2020, 8:20pm UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765 "2020-05-01T20:20:23Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![rishabh\_jain](https://avatars.discourse-cdn.com/v4/letter/r/aeb1de/32.png) [@rishabh\_jain](https://discuss.elastic.co/u/rishabh_jain)\
**Post date:** [May 1, 2020, 8:20pm UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765/1 "2020-05-01T20:20:23Z")

</div>

is there any way by which the data sent into kafka stream can be dumped into elasticsearch?

---

<div class="post-metadata">

**Author:** ![Luca\_Belluccini](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/luca_belluccini/32/33239_2.png) [@Luca\_Belluccini](https://discuss.elastic.co/u/Luca_Belluccini)\
**Post date:** [May 1, 2020, 8:46pm UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765/2 "2020-05-01T20:46:11Z")

</div>

Hello @rishabh_jain

Not directly... You need something in the middle.

Few options:

- [ES-Hadoop for Spark](https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html)
- [Logstash with Kafka Input](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html)
- [Filebeat with Kafka Input](https://www.elastic.co/guide/en/beats/filebeat/master/filebeat-input-kafka.html)
- [Kafka Connector for Elasticsearch](https://github.com/confluentinc/kafka-connect-elasticsearch) - Not by Elastic

---

<div class="post-metadata">

**Author:** ![rishabh\_jain](https://avatars.discourse-cdn.com/v4/letter/r/aeb1de/32.png) [@rishabh\_jain](https://discuss.elastic.co/u/rishabh_jain)\
**Post date:** [May 2, 2020, 10:05am UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765/3 "2020-05-02T10:05:43Z")

</div>

Thanks.  
Can you help me with one more problem  
is there any way in which i can do Upsert on the basis of version  
i.e if document doesn't exists then it would create the document  
and if exists then it would look for version if new version is greater than existing version then only it would update the document.

---

<div class="post-metadata">

**Author:** ![Christian\_Dahlqvist](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/christian_dahlqvist/32/4617_2.png) [@Christian\_Dahlqvist](https://discuss.elastic.co/u/Christian_Dahlqvist)\
**Post date:** [May 2, 2020, 10:09am UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765/4 "2020-05-02T10:09:17Z")

</div>

You should be able to use the [version and version\_type request parameters](https://www.elastic.co/guide/en/elasticsearch/reference/7.6/docs-index_.html#docs-index_) to control this. Not sure which ingest methods do or do not support this though.

---

<div class="post-metadata">

**Author:** ![rishabh\_jain](https://avatars.discourse-cdn.com/v4/letter/r/aeb1de/32.png) [@rishabh\_jain](https://discuss.elastic.co/u/rishabh_jain)\
**Post date:** [May 2, 2020, 7:31pm UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765/5 "2020-05-02T19:31:55Z")

</div>

POST API does not support external versioninig is there any way ?

---

<div class="post-metadata">

**Author:** ![Luca\_Belluccini](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/luca_belluccini/32/33239_2.png) [@Luca\_Belluccini](https://discuss.elastic.co/u/Luca_Belluccini)\
**Post date:** [May 2, 2020, 11:26pm UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765/6 "2020-05-02T23:26:10Z")

</div>

Hello @rishabh_jain

By Update you mean partially updating a document or overwriting the existing one?

---

<div class="post-metadata">

**Author:** ![rishabh\_jain](https://avatars.discourse-cdn.com/v4/letter/r/aeb1de/32.png) [@rishabh\_jain](https://discuss.elastic.co/u/rishabh_jain)\
**Post date:** [May 3, 2020, 12:33pm UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765/7 "2020-05-03T12:33:06Z")

</div>

want to partially update a document.  
and also if the document doesn't exists then it should be created

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [May 31, 2020, 12:33pm UTC](https://discuss.elastic.co/t/kafka-consumer-elasticsearch/230765/8 "2020-05-31T12:33:24Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.