title: How to connect Kafka with Elasticsearch?
body: <p>I am new in Kafka, I use kafka to collect netflow through logstash(it is ok), and I want to send the data to elasticsearch from kafka, but there are some problems.<br>
  My question is how can I connect Kafka with Elasticsearch?
  netflow to kafka logstash config:</p>
  
  <pre><code>input{
      udp{
          host =&gt; "120.127.XXX.XX"
          port =&gt; 5556
          codec =&gt; netflow
      }
  }
      filter{
  
      }
  output {
    kafka {
      bootstrap_servers =&gt; "localhost:9092"    
      topic_id =&gt; "test"    
    }
    stdout{codec=&gt; rubydebug}
  }
  </code></pre>
  
  <p>kafka to elasticsearch logstash:</p>
  
  <pre><code>input {
        kafka { }
      }
      output {
          elasticsearch {
              hosts =&gt; ["120.127.XXX.XX:9200"]
          }
          stdout{codec=&gt; rubydebug}
      }
  </code></pre>
  
  <p>log:</p>
  
  <pre><code>D:\ELK\logstash-6.1.1\bin&gt;logstash -f kafkatoES.conf --path.data D:\ELK\logstash-6.1.1\datatest
  Sending Logstash's logs to D:/ELK/logstash-6.1.1/logs which is now configured via log4j2.properties
  [2018-02-01T18:52:59,713][INFO ][logstash.modules.scaffold] Initializing module {:module_name=&gt;"fb_apache", :directory=&gt;"D:/ELK/logstash-6.1.1/modules/fb_apache/configuration"}
  [2018-02-01T18:52:59,728][INFO ][logstash.modules.scaffold] Initializing module {:module_name=&gt;"netflow", :directory=&gt;"D:/ELK/logstash-6.1.1/modules/netflow/configuration"}
  [2018-02-01T18:53:00,072][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
  [2018-02-01T18:53:01,070][INFO ][logstash.runner ] Starting Logstash {"logstash.version"=&gt;"6.1.1"}
  [2018-02-01T18:53:01,804][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=&gt;9601}
  [2018-02-01T18:53:09,024][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=&gt;{:removed=&gt;[], :added=&gt;[http://120.127.XX.XX:9200/]}}
  [2018-02-01T18:53:09,040][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=&gt;http://120.127.XX.XX:9200/, :path=&gt;"/"}
  [2018-02-01T18:53:09,305][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=&gt;"http://120.127.XX.XX:9200/"}
  [2018-02-01T18:53:09,383][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=&gt;nil}
  [2018-02-01T18:53:09,383][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the type event field won't be used to determine the document _type {:es_version=&gt;6}
  [2018-02-01T18:53:09,415][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=&gt;nil}
  [2018-02-01T18:53:09,430][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=&gt;{"template"=&gt;"logstash-", "version"=&gt;60001, "settings"=&gt;{"index.refresh_interval"=&gt;"5s"}, "mappings"=&gt;{"default"=&gt;{"dynamic_templates"=&gt;[{"message_field"=&gt;{"path_match"=&gt;"message", "match_mapping_type"=&gt;"string", "mapping"=&gt;{"type"=&gt;"text", "norms"=&gt;false}}}, {"string_fields"=&gt;{"match"=&gt;"", "match_mapping_type"=&gt;"string", "mapping"=&gt;{"type"=&gt;"text", "norms"=&gt;false, "fields"=&gt;{"keyword"=&gt;{"type"=&gt;"keyword", "ignore_above"=&gt;256}}}}}], "properties"=&gt;{"@timestamp"=&gt;{"type"=&gt;"date"}, "@version"=&gt;{"type"=&gt;"keyword"}, "geoip"=&gt;{"dynamic"=&gt;true, "properties"=&gt;{"ip"=&gt;{"type"=&gt;"ip"}, "location"=&gt;{"type"=&gt;"geo_point"}, "latitude"=&gt;{"type"=&gt;"half_float"}, "longitude"=&gt;{"type"=&gt;"half_float"}}}}}}}}
  [2018-02-01T18:53:09,493][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=&gt;"LogStash::Outputs::ElasticSearch", :hosts=&gt;["//120.127.XXX.XX:9200"]}
  [2018-02-01T18:53:09,524][INFO ][logstash.pipeline ] Starting pipeline {:pipeline_id=&gt;"main", "pipeline.workers"=&gt;16, "pipeline.batch.size"=&gt;125, "pipeline.batch.delay"=&gt;5, "pipeline.max_inflight"=&gt;2000, :thread=&gt;"#&lt;Thread:0x45e62903 run&gt;"}
  [2018-02-01T18:53:09,609][INFO ][logstash.pipeline ] Pipeline started {"pipeline.id"=&gt;"main"}
  SLF4J: Class path contains multiple SLF4J bindings.
  SLF4J: Found binding in [jar:file:/D:/ELK/logstash-6.1.1/logstash-core/lib/org/apache/logging/log4j/log4j-slf4j-impl/2.6.2/log4j-slf4j-impl-2.6.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]
  SLF4J: Found binding in [jar:file:/D:/ELK/logstash-6.1.1/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.0.2/vendor/jar-dependencies/runtime-jars/log4j-slf4j-impl-2.8.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]
  SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
  SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
  [2018-02-01T18:53:09,789][INFO ][logstash.agent ] Pipelines running {:count=&gt;1, :pipelines=&gt;["main"]}
  [2018-02-01T18:53:09,852][INFO ][org.apache.kafka.clients.consumer.ConsumerConfig] ConsumerConfig values:
  auto.commit.interval.ms = 5000
  auto.offset.reset = latest
  bootstrap.servers = [localhost:9092]
  check.crcs = true
  client.id = logstash-0
  connections.max.idle.ms = 540000
  enable.auto.commit = true
  exclude.internal.topics = true
  metrics.recording.level = INFO
  metrics.sample.window.ms = 30000
  partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
  receive.buffer.bytes = 65536
  reconnect.backoff.max.ms = 1000
  reconnect.backoff.ms = 50
  request.timeout.ms = 305000
  retry.backoff.ms = 100
  sasl.jaas.config = null
  sasl.kerberos.kinit.cmd = /usr/bin/kinit
  ssl.keystore.password = null
  ssl.keystore.type = JKS
  ssl.protocol = TLS
  ssl.provider = null
  ssl.secure.random.implementation = null
  ssl.trustmanager.algorithm = PKIX
  ssl.truststore.location = null
  ssl.truststore.password = null
  ssl.truststore.type = JKS
  value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
  
  [2018-02-01T18:53:09,945][INFO ][org.apache.kafka.common.utils.AppInfoParser] Kafka version : 0.11.0.0
  [2018-02-01T18:53:09,945][INFO ][org.apache.kafka.common.utils.AppInfoParser] Kafka commitId : cb8625948210849f
  [2018-02-01T18:53:10,149][INFO ][org.apache.kafka.clients.consumer.internals.AbstractCoordinator] Discovered coordinator winoc-netflow:9092 (id: 2147483647 rack: null) for group logstash.
  [2018-02-01T18:53:10,164][INFO ][org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] Revoking previously assigned partitions [] for group logstash
  [2018-02-01T18:53:10,164][INFO ][org.apache.kafka.clients.consumer.internals.AbstractCoordinator] (Re-)joining group logstash
  [2018-02-01T18:53:10,180][INFO ][org.apache.kafka.clients.consumer.internals.AbstractCoordinator] Successfully joined group logstash with generation 6
  [2018-02-01T18:53:10,180][INFO ][org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] Setting newly assigned partitions [logstash-0] for group logstash
  </code></pre>
  
  <p>thank you in advance!</p>