# How to pull data from kafka to elasticsearch

**URL:** <https://discuss.elastic.co/t/how-to-pull-data-from-kafka-to-elasticsearch/173445>\
**Category:** Logstash\
**Created:** [March 22, 2019, 7:36am UTC](https://discuss.elastic.co/t/how-to-pull-data-from-kafka-to-elasticsearch/173445 "2019-03-22T07:36:32Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![khergner](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/khergner/32/130758_2.png) [@khergner](https://discuss.elastic.co/u/khergner)\
**Post date:** [March 22, 2019, 7:36am UTC](https://discuss.elastic.co/t/how-to-pull-data-from-kafka-to-elasticsearch/173445/1 "2019-03-22T07:36:33Z")

</div>

Hi everyone  
I want to send data from beat to kafka after kafka send logstash. Kafka role is message que here. I would like to use two beat but not created index based on elastic.

Logstash

```
input {
  kafka {
	 bootstrap_servers => "172.28.26.169:9092"
     topics => ["metricbeat"]
	 codec => "json"
  }
}

filter {
  if [@metadata][kafka][topic] == "metricbeat" {
    mutate {
      add_field => { "[es_index]" => "metricbeat2-%{+YYYY.MM.dd}"}
    }
  } else if [@metadata][kafka][topic] == "filebeat" {
    mutate {
      add_field => { "[es_index]" => "filebeat"}
    }
  }
}

output {
  elasticsearch {
    hosts => ["172.28.26.169:9200"]
	index => "%{es_index}"
	workers => 1
  }
}
```

---

<div class="post-metadata">

**Author:** ![A\_B](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/a_b/32/17104_2.png) [@A\_B](https://discuss.elastic.co/u/A_B)\
**Post date:** [March 22, 2019, 11:14am UTC](https://discuss.elastic.co/t/how-to-pull-data-from-kafka-to-elasticsearch/173445/2 "2019-03-22T11:14:31Z")

</div>

Hi @khergner,

I do not use the Kafka Logstash input module but as far as I can tell that config looks good. What problems do you have or are you asking advice before implementing this at all?

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [March 22, 2019, 12:53pm UTC](https://discuss.elastic.co/t/how-to-pull-data-from-kafka-to-elasticsearch/173445/3 "2019-03-22T12:53:22Z")

</div>

You need to set [decorate\_events](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-decorate_events) =\> true on the kafka input.

---

<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:** [April 19, 2019, 12:53pm UTC](https://discuss.elastic.co/t/how-to-pull-data-from-kafka-to-elasticsearch/173445/4 "2019-04-19T12:53:23Z")

</div>

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