视频地址

B 站视频地址：Logstash 如何成为镇得住局面的数据管道

公众号视频地址：Logstash 如何成为镇得住局面的数据管道

知乎视频地址：Logstash 如何成为镇得住局面的数据管道

内容

首先咱们连续上一期视频中日志采集架构的案例，Filebeat 采集日志并推送 Kafka 音讯队列进行散发，再由 Logstash 生产日志音讯，并将日志数据最终落地在 Elasticsearch 集群索引当中，Kafka 作为音讯队列散发服务须要将收集到的日志音讯持续散发上来，最终数据落地在 Elasticsearch 集群索引当中。

那么连贯整个过程的配角 Logstash 是如何工作的，就是咱们明天解说的重点。

Logstash 工作过程分为三个局部：Input 输出、Filter 过滤、Output 输入，它们独特合作造成了残缺的 Logstash 数据管道传输机制

咱们先从一个最简略的例子演示开始，看看 Logstash 是怎么输出和输入的，这一次先跳过 filter 过滤环节。

上面查看曾经预置好的一个配置文件 01-kafka-elastic-nginx.conf

首先是 input 输出配置点，从 Kafka 订阅音讯，Kafka 集群地址与 filebeat 中都指向了一个地址，其余配置咱们先略过，后续 Kafka 专题再说

下来看到要订阅的 Topic 主题 TestT3，咱们先不必 json 格局解码音讯，默认就是纯文本的形式

一样的，这一步先略过过滤环节，间接看看 output 输入配置点，指标是给 Elasticsearch 输入数据，并指定了 elasticsearch 集群的三个节点

输入环节创立须要写入的 elasticsearch 日志索引，咱们先依照默认的 filebeat 采集工夫，进行日期格式化，依照每个小时建设一个索引，这块会有工夫问题，一会儿再说。

让数据输入到终端，不便咱们调试后果。

通过演示中最简略的配置形式，这时候的 Logstash 曾经成为连贯 Kafka 和 Elastisearch 之间的数据管道了！

好，接下来咱们将所有零碎运行起来，并生成一条 nginx 申请日志，看看管道各个阶段的数据变动。

首先 nginx 日志数据被 filebeat 采集，是一条典型的无构造的文本日志数据，大家留神红色标注的工夫是 2021 年 2 月 21 日 13 时

接着这条日志数据通过 Kafka 进入到了 Logstash 管道的输出阶段，

Logstash 为这条日志生成了更为十分宏大的 Json 数据，外面包含了所有被采集主机的信息，以及 nginx 日志，实际上这些原始信息并没有被良好的进行数据荡涤与结构化

最初数据被写入到 Elastisearch 一个按小时划分的索引当中，对应工夫为 2021 年 2 月 21 日 5 时

咱们发现 Logstash 对原始数据在没有任何解决的状况下，会很不不便未来数据的应用；

这次咱们利用 Logstash json 解码器让管道从新再来一次，

接下来咱们进入 Logstash 中对应的配置文件，并找到 input 输出点的 codec 配置，删掉正文，关上 Logstash 对输出数据的 json 解码形式·。

咱们看看再次进入管道中的日志数据，Logstash 首先对原始日志数据进行 Json 解析

这时候咱们再看 Json 解析后的数据，是不是就清晰多了，filebeat 采集到的本地机器数据、以及红色框中 Nginx HTTP 日志数据、以及其余标签数据都进行了字段拆散

做到这一步其实还是不够好，为什么呢？一方面因为咱们仍然心愿将 Nginx HTTP 的日志数据也进行结构化解决，

另一个方面，Filebeat 传递给 Logstash 的零碎工夫是慢了 8 个小时的 UTC 工夫规范，反而 Nginx 日志中的工夫是咱们本地的北京工夫规范，因而咱们心愿用 Nginx 日志工夫作为创立 Elasticsearch 日志索引的惟一根据

这时候咱们就要应用 Logstash 的过滤机制了，咱们持续进入 Logstash 对应的配置中，删掉过滤配置中的正文，让 Logstash 过滤最罕用插件 grok、date、ruby、mutate 起作用

grok 插件是业余解决非结构化数据的能手，通过自定义的 Nginx 日志正则表达式，就能实现 Nginx 日志的结构化解析

date 插件用于解决工夫问题，咱们通过 date 插件将 nginx 日志中的工夫转换成 Logstash 工夫对象，并赋给一个新的长期工夫字段 indextime

ruby 就是在过滤过程中能够插入 ruby 脚本语言来进行程序级解决，咱们通过 ruby 语言对 indextime 工夫格式化，生成一个准确到小时的字符串字段 index.date，用于 elasticsearch 索引名称

mutate 是最罕用的能够对管道中数据字段进行操作的插件了，咱们的目标是删除长期工夫字段 indextime

最初咱们还须要将 output 输入中的索引生成形式批改一下，正文掉原来用 filebeat 生默认工夫生成的索引，改成 nginx 日志工夫生成的索引。

咱们从新运行 Logstash，数据通过了 Input 解码、日志 grok 结构化解决、本地工夫对象创立，并进行日期格式化，为了生成新的 Elasticsearch 索引字段，并对长期字段进行删除，最终通过 Output 输入阶段，创立 Elasticsearch 索引或写入日志数据

让咱们看看 Elasticsearch 最终保留的数据成果，index 索引对应的工夫来自过滤器创立的 index.date 字段，index.date 字段又来自 nginx 日志中拆散出的本地工夫。这样咱们就不必再去批改 Logstash 的零碎工夫了

咱们看到菱形标注的字段数据就是由过滤器对 nginx http 日志进行结构化抽取的后果，

同样 elasticsearch 仍然保留着 nginx 日志的原始数据以备不时之需

返回读字节的知乎——理解更多对于大数据的常识

公众号“读字节”分布式，大数据，软件架构的深度，业余解读