项目通过log4j2把日志写到了Kafka中，为了进一步分析数据通过logstash取出kafka的数据，经过filter处理之后，存入到elasticsearch中。

log4j2写入kafka主要是配置logj2.xml文件，加入kafka的配置和日志输出。

主题要配置正确，ip和端口号要配置kafka的，不是zookeeper的。Loggers要加入配好的AUDIT_KAFKA.

logstash使用的是6.2.3，首先是input输入，logstash集成了kafka的插件，直接用就成了。配置kafka的bootstrap地址。格式用json格式。topic一定要配置正确。

然后是filter部分，因为取的审计日志，特征比较明显，直接用split进行分割。

这里用mutate插件，并且将分割好的数据，以自己需要的名称来命名。还可以进行二次分割，再写一个mutate就好。

最后是输出部分，输出到elasticsearch，也是直接写elasticsearch就好。

就这样分割完了。

之后需要用的数据，可以通过java端写代码来获取。这部分还是之后再写吧。

学习过程中找到一篇elk的资料，和elasticsearch header的windows版本的安装。地址：https://download.csdn.net/download/qq_28600087/10820880