目前要把kafka中的数据传输到elasticsearch大概有以下几种方法：

1) logstash

2) flume

3) spark streaming

4) kafka connect

5) 开发程序消费kafka写入elasticsearch

本文介绍如何使用Logstash将Kafka中的数据写入到ElasticSearch，这里Kafka、logstash、elasticsearch安装就详述了。

Logstash工作的流程由三部分组成：

input：输入（即source），表示从那里采集数据

filter：过滤，logstash对数据的ETL就是在这个里面进行。

output：输出（即sink），表示数据输出地方。

注意：input需要logstash-input-kafka插件，该插件logstash默认自带。

1) input输入

说明：

decorate_events：此属性会将当前topic、offset、group、partition等信息也带到message中，可以达到订阅多个主题在es中为不同主题创建不同的索引。

codec => "json"：表示会将消息格式为json，如果不加这个参数，整条数据变成一个字符串存储到message字段里面。

2) filter过滤

说明：根据业务需求进行ETL数据处理。

这里，我为每个主题构建对应的[@metadata][index]，并在接下来output中引用。

3) output输出

场景：

消费kafka中多个主题数据(json格式)，通过logstash采集并根据不同主题输出到Elasticsearch中不同的索引中。

运行测试(这里，我为了方便测试打开了stdout输出到屏幕)：

主题test01对应的索引为 kafka-test01-2020.05.08

数据格式： 

主题test02对应的索引为 kafka-test02-2020.05.08 

数据格式： 

 最后，我们可以在head插件中查看输出的数据。