部署ELK+Kafka+Filebeat日志收集分析系统-Toy模板网

这篇具有很好参考价值的文章主要介绍了部署ELK+Kafka+Filebeat日志收集分析系统。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

部署ELK+Kafka+Filebeat日志收集分析系统

一、ELK 简介

ELK是三个软件的统称，即Elasticsearch、Logstash和Kibana三个开源软件的缩写。这三款软件都是开源软件，通常配合使用，并且都先后归于Elastic.co企业名下，故被简称为ELK协议栈。ELK主要用于部署在企业架构中，收集多台设备上多个服务的日志信息，并将其统一整合后提供给用户。它可以从任何来源、任何格式进行日志搜索、分析与可视化展示。

1、ELK日志分析系统组成

在ELK架构中，Elasticsearch、Logstash和Kibana三款软件作用如下：

elasticsearch (es) :通过搭建群集;存储日志数据，索引日志数据
logstash :收集日志，收集到了后给es存储
kibana :视图形式展现日志信息，更加人性化

2、Elasticsearch（es）

Elasticsearch是一个高度可扩展的全文搜索和分析引擎，基于Apache Lucence（事实上，Lucence也是百度所采用的搜索引擎）构建，能够对大容量的数据进行接近实时的存储、搜索和分析操作。

3、Logstash

Logstash是一个数据收集引擎，它可以动态的从各种数据源搜集数据，并对数据进行过滤、分析和统一格式等操作，并将输出结果存储到指定位置上。Logstash支持普通的日志文件和自定义Json格式的日志解析。

4、Kibana

Kibana是一个数据分析和可视化平台，通常与Elasticsearch配合使用，用于对其中的数据进行搜索、分析，并且以统计图标的形式展示。

5、日志处理步骤

1.将APP servers的日志进行集中化管理到Logstash agent。
2.将日志格式化（Logstash）并输出到Elasticsearch cluster
3.对格式化后的数据进行索引和存储（Elasticsearch）
4.前端数据的展示（Kibana）
5.可以在线查看界面化展示。

ELK是由Elasticsearch、Logstash、Kiban三个开源软件的组合。
Logstash 收集APP server产生的log，然后存放到Elasticsearch集群节点中
kibana从Elasticsearch集群节点中查询数据生成图表，再返回给Brower。

二、Elasticsearch介绍

提供了一个分布式多用户能力的全文搜索引擎；
是一个基于Lucene的搜索服务器；
基于restful web接口；
使用java开发；
作为apache许可条款下的开放源码发布，是第二流行的企业搜索引擎；
被设计用于云计算中，能够达到实时搜索、稳定、可靠、快速、安装实用方便的需求。

1、Elasticsearch核心概念

接近实时(NRT)
elasticsearch是一个接近实时的搜索平台，这意味着，从索引一个文档直到这个文档能够被搜索到有一个轻微的延迟（通常是1秒)

cluster集群，ES是一个分布式的系统

一个集群就是由一个或多个节点组织在一起，它们共同持有你整个的数据，并一起提供索引和搜索功能。其中一个节点为主节点，这个主节点是可以通过选举产生的，并提供跨节点的联合索引和搜索的功能。集群有一个唯一性标示的名字，默认是elasticsearch，集群名字很重要，每个节点是基于集群名字加入到其集群中的。因此，确保在不同环境中使用不同的集群名字。
—个集群可以只有一个节点。强烈建议在配置elasticsearch时，配置成集群模式。es具有集群机制，节点通过集群名称加入到集群中，同时在集群中的节点会有一个自己的唯一身份标识（自己的名称)
Node节点，就是集群中的一台服务器
节点就是一台单一的服务器，是集群的一部分，存储数据并参与集群的索引和搜索功能。像集群一样，节点也是通过名字来标识，默认是在节点启动时随机分配的字符名。当然，你可以自己定义。该名字也很重要，在集群中用于识别服务器对应的节点。
节点可以通过指定集群名字来加入到集群中。默认情况，每个节点被设置成加入到elasticsearch集群。如果启动了多个节点，假设能自动发现对方，他们将会自动组建一个名为elasticsearch的集群。
index索引
一个索引就是一个拥有几分相似特征的文档的集合。比如说，你可以有一个客户数据的索引、一个产品目录的索引、还有一个订单数据的索引。一个索引用一个名字来标识（必须全部是小写字母组合），并且当我们要对相应的索引中的文档进行索引、收缩、更新和删除的时候，都要用到这个名字。在一个集群中，可以定义多个索引。（索引相对于关系型数据库的库）

类型相对于关系型数据库的表 ——》索引(库)-》类型(表)-》文档(记录)

类型（type）
类型（type）在一个索引中，你可以定义一种或多种类型。一个类型是你的索引的一个逻辑上的分类分区，其寓意完全由你来定义。通常，会为具有一组共同字段的文档定义一个类型。比如：我们假设运营一个博客平台并且将所有的数据存储到一个索引中，在这个索引中，你可以为用户数据定义一个类型，为博客数据定义一个类型，也可以为评论数据定义另一个类型。（类型相对于关系型数据库的表）

分片和副本(shards & replicas)
在实际情况下，索引存储的数据可能超过单个节点的硬件限制。如一个10亿文档需1TB空间可能不适合存储在单个节点的磁盘上或者从单个节点搜索请求太慢了。为了解决这个问题，elasticsearch提供将索引分成多个分片的功能。当在创建索引时，可以定义想要分片的数量。每一个分片就是一个全功能的独立的索引，可以位于集群中任何节点上。

2、开启分片副本的主要原因

Elasticsearch采用分片式，可以进行水平分割横向扩展，增大存储量；分布式并行跨分片操作，提高性能和吞吐量。

高可用性，以应对分片或者节点故障，处于这个原因，分片副本要在不同节点上；
提高I/O性能，增大吞吐量，搜索可以并行在所有副本执行。总之，每个索引可以被分成多个分片，一个索引也可以被复制0次或者多次。一旦复制了，每个索引就有了主分片（可以作为复制源的原始分片）和复制分片（主分片的拷贝）之分。分片和副本的数量可以在索引创建的时候指定，在索引创建之后，你可以在任何时候动态改变副本的数量，但是你事后无法改变分片的数量。默认情况下，Elasticsearch中的每个索引被分片为5个主分片和1个副本，这意味着，如果你的集群中至少有两个节点的情况下，你的索引将会有5个主分片和另外5个副本分片（1个完全拷贝），这样的话每个索引总共就有10个分片。

三、Logstash

1.—款强大的数据处理工具

2.可实现数据传输、格式处理、格式化输出

3.数据输入(从业务输入)、数据加工(如过滤、改写等）以及数据输出（输出到Elasticsearch群集)

4.由LRuby语言编写，基于消息（message-based）的简单架构，并运行在Java虚拟机（JVM）上；

5.不同于分离的代理端（agent）或主机端（server），Logstash可配置单一的代理端（agent）与其他开源软件结合，以实现不同的功能。

1、Logstash的主要组件

Shipper：日志收集负责监控本地日志文件的变化，及时把日志文件的最新内容收集起来。通常，远程代理端（agent）只需要运行这个组件即可。
Indexer：日志存储负责接受日志并写入到本地文件。
Broker：日志hub负责链接多个shipper和对应数目的indexer。
Search and Storage允许对事件进行搜索和存储。
Web Interface基于web的展示界面。以上组件在Logstash架构中可以独立部署，因此提供了很好的集群扩展性

四、Kibana介绍

1、Kibana概述

一个针对Elasticsearch的开源分析及可视化平台；
搜索、查看存储在Elasticsearch索引中的数据；
通过各种图标进行高级数据分析及展示；
让海量数据更容易理解；
操作简单，基于浏览器地用户界面就可以快速创建仪表板（dashboard）实时显示Elasticsearch查询动态；
设置安装Kibana非常简单，无需编写代码，几分钟内就可以完成Kibana安装并启动Elasticsearch监测。

2、Kibana主要功能

Elasticsearch无缝之集成。Kibana架构为Elasticsearch定制，可以将任何结构化和非结构化数据加入Elasticsearch索引。Kibana还充分利用了Elasticsearch强大的搜索和分析功能。

整合数据：Kibana能够更好地处理海量数据，并据此创建柱形图、折线图、散点图、直方图、饼图和地图。

让更多团队成员受益：强大的数据库可视化接口让各业务岗位都能够从数据集合受益。

接口灵活，分享更容易：使用Kibana可以更加方便地创建、保存、分享数据，并将可视化数据快速交流。

配置简单：Kibana的配置和启用非常简单，用户体验非常友好。Kibana自带Web服务器，可以快速启动运行。

**可视化多数据源*8：Kibana可以非常方便地把来自Logstash、ES-Hadoop、Beats或第三方技术的数据整合到Elasticsearch，支持的第三方技术包括Apache Flume、Fluentd等。

简单数据导出：Kibana可以方便地导出感兴趣的数据，与其它数据集合并融合后快速建模分析，发现新结果。

五、ELK优点

处理方式灵活。 Elasticsearch是全文索引，具有强大的搜索能力。
2.配置相对简单。 Kibana的配置非常简单，Elasticsearch则全部使用Json接口，配置也不复杂，Logstash的配置使用模块的方式，配置也相对简单。
3.检索性能高。 ELK架构通常可以达到百亿级数据的查询秒级响应。
4.集群线性扩展。 Elasticsearch本身没有单点的概念，自动默认集群模式，Elasticsearch和Logstash都可以
5.灵活扩展。
6.页面美观。 Kibana的前端设计美观，且操作简单。

六、Filebeat 简介

Filebeat由两个主要组成部分组成：prospector(探勘者)和 harvesters(矿车)。这些组件一起工作来读取文件并将事件数据发送到指定的output。

1、prospector: 负责找到所有需要进行读取的数据源

2、harvesters：负责读取单个文件的内容，并将内容发送到output中，负责文件的打开和关闭。

1、Logstash

作为数据收集引擎。它支持动态的从各种数据源搜集数据，并对数据进行过滤、分析、丰富、统一格式等操作，然后存储到用户指定的位置,一般会发送给 Elasticsearch。

可以添加的其它组件：

2、Filebeat

轻量级的开源日志文件数据搜集器。通常在需要采集数据的客户端安装 Filebeat，并指定目录与日志格式，Filebeat 就能快速收集数据，并发送给 logstash 进行解析，或是直接发给 Elasticsearch 存储，性能上相比运行于 JVM 上的 logstash 优势明显，是对它的替代。常应用于 EFLK 架构当中。

3、Filebeat的工作方式

启动Filebeat时，它将启动一个或多个输入，这些输入将在为日志数据指定的位置中查找。对于Filebeat所找到的每个日志，Filebeat都会启动收集器。每个收集器都读取单个日志以获取新内容，并将新日志数据发送到libbeat，libbeat将聚集事件，并将聚集的数据发送到为Filebeat配置的输出。

4、Filebeat工作原理

Filebeat可以保持每个文件的状态，并且频繁地把文件状态从注册表里更新到磁盘。这里所说的文件状态是用来记录上一次Harvster读取文件时读取到的位置，以保证能把全部的日志数据都读取出来，然后发送给output。如果在某一时刻，作为output的ElasticSearch或者Logstash变成了不可用，Filebeat将会把最后的文件读取位置保存下来，直到output重新可用的时候，快速地恢复文件数据的读取。在Filebaet运行过程中，每个Prospector的状态信息都会保存在内存里。如果Filebeat出行了重启，完成重启之后，会从注册表文件里恢复重启之前的状态信息，让FIlebeat继续从之前已知的位置开始进行数据读取。

5、Filebeat用途

适用于集群环境下，服务多，且部署在不同机器
1、为什么要用filebeat来收集日志？
因为logstash是jvm跑的，资源消耗比较大，启动一个logstash就需要消耗500M左右的内存（这就是为什么logstash启动特别慢的原因），而filebeat只需要10来M内存资源。常用的ELK日志采集方案中，大部分的做法就是将所有节点的日志内容通过filebeat发送到logstash，logstash根据配置文件进行过滤。然后将过滤之后的文件输送到elasticsearch中，通过kibana去展示。

2、filebeat结合logstash带来好处

通过 Logstash 具有基于磁盘的自适应缓冲系统，该系统将吸收传入的吞吐量，从而减轻 Elasticsearch 持续写入数据的压力
从其他数据源（例如数据库，S3对象存储或消息传递队列）中提取
将数据发送到多个目的地，例如S3，HDFS（Hadoop分布式文件系统）或写入文件
使用条件数据流逻辑组成更复杂的处理管道
缓存/消息队列（redis、kafka、RabbitMQ等）：可以对高并发日志数据进行流量削峰和缓冲，这样的缓冲可以一定程度的保护数据不丢失，还可以对整个架构进行应用解耦。

Fluentd：是一个流行的开源数据收集器。由于 logstash 太重量级的缺点，Logstash 性能低、资源消耗比较多等问题，随后就有 Fluentd 的出现。相比较 logstash，Fluentd 更易用、资源消耗更少、性能更高，在数据处理上更高效可靠，受到企业欢迎，成为 logstash 的一种替代方案，常应用于 EFK 架构当中。在 Kubernetes 集群中也常使用 EFK 作为日志数据收集的方案。

在 Kubernetes 集群中一般是通过 DaemonSet 来运行 Fluentd，以便它在每个 Kubernetes 工作节点上都可以运行一个 Pod。它通过获取容器日志文件、过滤和转换日志数据，然后将数据传递到 Elasticsearch 集群，在该集群中对其进行索引和存储。

3、Filebeat和Logstash的区别

	Logstash	filebeat
内存	大	小
cpu	大	小
插件	多	多
功能	从多种输入端采集并实时解析和转换数据并输出到多种输出端	传输
轻重	相对较重	轻量级二进制文件
过滤能力	强大的过滤能力	有过滤能力但是弱
集群	单节点	单节点
输出到多个接收方	支持	6.0之前支持
二次开发或者扩展开发	难	易
进程	一台服务器只允许一个logstash进程，挂掉之后需要手动拉起

七、Kafka简介

Kafka是一种消息队列，主要用来处理大量数据状态下的消息队列，一般用来做日志的处理。既然是消息队列，那么Kafka也就拥有消息队列的相应的特性了。

可以在系统中起到“肖峰填谷”的作用，也可以用于异构、分布式系统中海量数据的异步化处理。

1、为什么需要消息队列？

主要原因是由于在高并发环境下，同步请求来不及处理，请求往往会发生阻塞。比如大量的请求并发访问数据库，导致行锁表锁，最后请求线程会堆积过多，从而触发 too many connection 错误，引发雪崩效应。

我们使用消息队列，通过异步处理请求，从而缓解系统的压力。消息队列常应用于异步处理，流量削峰，应用解耦，消息通讯等场景。

当前比较常见的 MQ 中间件有 ActiveMQ、RabbitMQ、RocketMQ、Kafka 等。

2、消息队列的好处

1、解耦合

耦合的状态表示当你实现某个功能的时候，是直接接入当前接口，而利用消息队列，可以将相应的消息发送到消息队列，这样的话，如果接口出了问题，将不会影响到当前的功能。

2、异步处理

异步处理替代了之前的同步处理，异步处理不需要让流程走完就返回结果，可以将消息发送到消息队列中，然后返回结果，剩下让其他业务处理接口从消息队列中拉取消费处理即可。

3、流量削峰

高流量的时候，使用消息队列作为中间件可以将流量的高峰保存在消息队列中，从而防止了系统的高请求，减轻服务器的请求处理压力。

3、Kafka的特性

高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒

可扩展性：kafka集群支持热扩展

持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失

容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）

高并发：支持数千个客户端同时读写

4、Kafka作为存储系统

1、任何允许发布与消费它们分离的消息的消息队列实际上充当了正在进行的消息的存储系统。

2、Kafka的不同之处在于它是一个非常好的存储系统。

3、写入Kafka的数据将写入磁盘并进行复制以实现容错。Kafka允许生产者等待确认，以便在完全复制之前写入不被认为是完整的，并且即使写入的服务器失败也保证写入仍然存在。

4、磁盘结构Kafka很好地使用了规模 - 无论服务器上有50 KB还是50 TB的持久数据，Kafka都会执行相同的操作。

八、Kafka消费模式

Kafka的消费模式主要有两种：

1、一种是一对一的消费，也即点对点的通信，即一个发送一个接收。

2、第二种为一对多的消费，即一个消息发送到消息队列，消费者根据消息队列的订阅拉取消息消费。

1、一对一

消息生产者发布消息到Queue（队列）队列中，通知消费者从队列中拉取消息进行消费。消息被消费之后则删除，Queue支持多个消费者，但对于一条消息而言，只有一个消费者可以消费，即一条消息只能被一个消费者消费。

2、一对多

这种模式也称为发布/订阅模式，即利用Topic（主题）存储消息，消息生产者将消息发布到Topic中，同时有多个消费者订阅此topic，消费者可以从中消费消息，注意发布到Topic中的消息会被多个消费者消费，消费者消费数据之后，数据不会被清除，Kafka会默认保留一段时间，然后再删除。

九、Kafka的基础架构

Kafka像其他Mq一样，也有自己的基础架构，主要存在生产者Producer、Kafka集群Broker、消费者Consumer、注册消息Zookeeper

Producer：Producer即生产者，消息的产生者，是消息的入口。

Broker：Broker是kafka实例，每个服务器上有一个或多个kafka的实例，我们姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号

Topic：消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。

Partition：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的文件夹！

Replication:每一个分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。

Message：每一条发送的消息主体。

Consumer：消费者，即消息的消费方，是消息的出口。

Consumer Group：我们可以将多个消费组组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！

*Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。

Leader：每个分区多个副本的主角色，生产者发送数据的对象，以及消费者消费数据的对象都是Leader。

Follower：每个分区多个副本的从角色，实时的从Leader中同步数据，保持和Leader数据的同步，Leader发生故障的时候，某个Follower会成为新的Leader。

简单来说：

上述一个Topic会产生多个分区Partition，分区中分为Leader和Follower，消息一般发送到Leader，Follower通过数据的同步与Leader保持同步，消费的话也是在Leader中发生消费，如果多个消费者，则分别消费Leader和各个Follower中的消息，当Leader发生故障的时候，某个Follower会成为主节点，此时会对齐消息的偏移量。

2、工作流程

producer就是生产者，是数据的入口。Producer在写入数据的时候永远的找leader，不会直接将数据写入follower

3、分区的原因

1、便在集群中扩展，每个Partition（分区）可以通过调整以适应它所在的机器，而一个topic（消息主题）又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；

2、可以提高并发，因为可以以Partition为单位读写了。

4、分区目的

producer（生产者）采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的。

数据会写入到不同的分区，分区的目的是

1、方便扩展：因为一个topic可以有多个partition，所以我们可以通过扩展机器去轻松的应对日益增长的数据量。

2、抗高并发：以partition为读写单位，可以多个消费者同时消费数据，提高了消息的处理效率。

十、Kafka原则

类似于负载均衡，当我们向某个服务器发送请求的时候，服务端可能会对请求做一个负载，将流量分发到不同的服务器，那在kafka中，如果某个topic有多个partition，producer又怎么知道该将数据发往哪个partition呢？kafka中有几个原则：

1、partition在写入的时候可以指定需要写入的partition，如果有指定，则写入对应的partition。

2、如果没有指定partition，但是设置了数据的key，则会根据key的值hash出一个partition。

3、如果既没指定partition，又没有设置key，则会轮询选出一个partition。

保证消息不丢失是一个消息队列中间件的基本保证，那producer在向kafka写入消息的时候，怎么保证消息不丢失呢？

那就是通过ACK应答机制！在生产者向队列写入数据的时候可以设置参数来确定是否确认kafka接收到数据，这个参数可设置的值为0、1、all。

十一、ZooKeeper简介

ZooKeeper是一种为分布式应用所设计的高可用、高性能且一致的开源协调服务，它提供了一项基本服务：分布式锁服务。分布式应用可以基于它实现更高级的服务，实现诸如同步服务、配置维护和集群管理或者命名的服务。

Zookeeper服务自身组成一个集群，2n+1个（奇数）服务允许n个失效，集群内一半以上机器可用，Zookeeper就可用。

假设 3台机器组成的集群，可以有允许一台失效，如果有2台失效，这个集群就不可用，1<1.5，一般的搭建zookeeper集群时，以奇数台机器来搭建。目的：是为了提高容错能允许多损失一台。

十二、 zookeeper工作机制

Zookeeper从设计模式角度来理解:是一个基于观察者模式设计的分布式服务管理框架它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。也就是说 Zookeeper =文件系统＋通知机制。

十三、zookeeper特点

1、Zookeeper:一个领导者(Leader)，多个跟随者(Follower）组成的集群。
 
2、Zookeepe集群中只要有半数以上节点存活，Zookeeper集群就能正常服务。所以zookeeper适合安装奇数台服务器。
 
3、全局数据一致:每个server保存一份相同的数据副本，client无论连接到哪个Server，数据都是一致的。
 
4、更新请求顺序执行，来自同一个client的更新请求按其发送顺序依次执行，即先进先出。
 
5、数据更新原子性，一次数据更新要么成功，要么失败。
 
6、实时性，在一定时间范围内，client能读到最新数据。]

十四、Zookeeper数据结构

ZooKeeper数据模型的结构与Linux文件系统很类似，整体上可以看作是一棵树，每个节点称做一个ZNode。每一个ZNode默认能够存储1MB的数据，每个ZNode都可以通过其路径唯一标识。

十五、zookeeper应用场景

提供的服务包括:统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。

1、统一命名服务

1、在分布式环境下，经常需要对应用/服务进行统一命名，便于识别。例如:IP不容易记住，而域名容易记住。

2、统一配置管理

1、分布式环境下，配置文件同步非常常见。一般要求一个集群中，所有节点的配置信息是一致的，比如Kafka集群。对配置文件修改后，希望能够快速同步到各个节点上。

2、配置管理可交由ZooKeeper实现。可将配置信息写入ZooKeeper上的一个Znode。各个客户端服务器监听这个Znode。一旦Znode中的数据被修改，ZooKeeper将通知各个客户端服务器。

3、统一集群管理

1、分布式环境中，实时掌握每个节点的状态是必要的。可根据节点实时状态做出一些调整。

2、ZooKeeper可以实现实时监控节点状态变化。可将节点信息写入ZooKeeper上的一个2Node。监听这个DMode可获取它的实时状态变化。

4、服务器动态上下线

客户端能实时洞察到服务器上下线的变化。

5、软负教均衡

在Zookeeper中记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户端请求。

十六、工作原理

Zookeeper的核心是原子广播，这个机制保证了各个Server之间的同步。实现这个机制的协议叫做Zab协议。
Zab协议有两种模式，它们分别是恢复模式（选主）和广播模式（同步）。

恢复模式：当服务启动或者在领导者崩溃后，Zab就进入了恢复模式，恢复模式不接受客户端请求，当领导者被选举出来，且大多数Server完成了和leader的状态同步以后，恢复模式就结束了。
状态同步保证了leader和Server具有相同的系统状态。

广播模式：一旦Leader已经和多数的Follower进行了状态同步后，他就可以开始广播消息了，

即进入广播状态。这时候当一个Server加入ZooKeeper服务中，它会在恢复模式下启动，发现Leader，并和Leader进行状态同步。待到同步结束，它也参与消息广播。

ZooKeeper的广播状态一直到Leader崩溃了或者Leader失去了大部分的Followers支持。

十七、zookeeper选举机制

第一次启动选举机制

1、服务器1启动，发起一次选举。服务器1投自己一票。
此时服务器1票数一票，不够半数以上（3票)，选举无法完成，服务器1状态保持为LOOKING;
 
2、服务器2启动，再发起一次选举。
服务器1和2分别投自己一票并交换选票信息:
此时服务器1发现服务器2的myid比自己目前投票推举的（服务器1)大，
更改选票为推举服务器2。此时服务器1票数0票，服务器2票数2票，没有半数以上结果，选举无法完成，服务器1，2状态保持LOOKING
 
3、服务器3启动，发起一次选举。此时服务器1和2都会更改选票为服务器3。
此次投票结果:服务器1为0票，服务器2为0票，服务器3为3票。此时服务器3的票数已经超过半数，
服务器3当选Leader。服务器1，2更改状态为FOLLOWING，服务器3更改状态为LEADING;
 
4、服务器4启动，发起一次选举。此时服务器1，2，3已经不是LooKING状态，不会更改选票信息。
交换选票信息结果:服务器3为3票，服务器4为1票。
此时服务器4服从多数，更改选票信息为服务器3，并更改状态为FOLOWING;
 
5、服务器5启动，同4一样当小弟。

非第一次启动选举机制

当ZooKeeper集群中的一台服务器出现以下两种情况之一时，就会开始进入Leader选举:
 
服务器初始化启动。
服务器运行期间无法和Leader保持连接。
而当一台机器进入Leader选举流程时，当前集群也可能会处于以下两种状态:
 
集群中本来就己经存在一个Leader。
对于已经存在Leader的情况，机器试图去选举Leader时，会被告知当前服务器的Leader信息，对于该机器来说，仅仅需要和Leader机器建立连接，并进行状态同步即可。
**集群中确实不存在Leader。**假设ZooKeeper由5台服务器组成，SID分别为1、2、3、4、5，ZXID分别为8、8、8、7、，并且此时sID为3的服务器是。一时刻，3和5服务器出现故障，因此开始进行Leader选举。
选举Leader规则:
 
EPOCH大的直接胜出
EPOCH相同，事务id大的胜出
事务id相同，服务器id大的胜出
 
 
SID:服务器ID。用来唯一标识一台ZooKeeper集群中的机器，每台机器不能重复，和myid一致。
ZXID:事务ID。ZXID是一个事务ID，用来标识一次服务器状态的变更。
在某一时刻，集群中的每台机器的zxID值不一定完全一致，这和ZooKeeper服务器对于客户端"更新请求"的处理逻辑速度有关。
Bpoch:每个Leader任期的代号。没有Leader时同一轮投票过程中的逻辑时钟值是相同的。每投完一次票这个数据就会增加

十八、ELK+Filebeat+Kafka+Zookeeper架构

部署ELK+Kafka+Filebeat日志收集分析系统,elk,kafka,分布式

每层实现的功能和含义分别介绍如下：
 
数据采集层
数据采集层位于最左边的业务服务器集群上，
在每个业务服务器上面安装了filebeat做日志收集，然后把采集到的原始日志发送到Kafka+zookeeper集群上。
 
消息队列层
原始日志发送到Kafka+zookeeper集群上后，会进行集中存储，
此时，filbeat是消息的生产者，存储的消息可以随时被消费。
 
数据分析层
Logstash作为消费者，会去Kafka+zookeeper集群节点实时拉取原始日志，
然后将获取到的原始日志根据规则进行分析、清洗、过滤，最后将清洗好的日志转发至Elasticsearch集群。
 
数据持久化存储
Elasticsearch集群在接收到logstash发送过来的数据后，
执行写磁盘，建索引库等操作，最后将结构化的数据存储到Elasticsearch集群上。
 
数据查询、展示层
Kibana是一个可视化的数据展示平台，当有数据检索请求时，
它从Elasticsearch集群上读取数据，然后进行可视化出图和多维度分析。

十九部署ELK+Filebeat+Kafka+Zookeeper

ip	部署的服务	主机名
192.168.0.102	es+kafka+zookeeper+kibana	elk1
192.168.0.103	es+kafka+zookeeper	elk2
192.168.0.104	kafka+zookeeper+nginx+logstash+filebeat	client

跑es集群的虚拟机要内存4G 处理器4核

//前期准备 关闭防火墙 selinux 修改主机名 安装java环境 配置域名解析
[root@localhost ~]# systemctl status firewalld
● firewalld.service - firewalld - dynamic firewall daemon
   Loaded: loaded (/usr/lib/systemd/system/firewalld.service; disabled; vendor preset: enabled)
   Active: inactive (dead)
     Docs: man:firewalld(1)
[root@localhost ~]# getenforce 
Disabled
[root@elk1 ~]# vi /etc/hosts 
[root@elk1 ~]# cat /etc/hosts 
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.0.102 elk1
192.168.0.103 elk2
192.168.0.104 client
[root@localhost ~]# hostnamectl set-hostname elk1
[root@localhost ~]# bash
[root@elk1 ~]# 
[root@elk1 ~]# yum -y install java
[root@elk1 ~]# java -version
openjdk version "1.8.0_312"
OpenJDK Runtime Environment (build 1.8.0_312-b07)
OpenJDK 64-Bit Server VM (build 25.312-b07, mixed mode)


[root@localhost ~]# systemctl status firewalld
● firewalld.service - firewalld - dynamic firewall daemon
   Loaded: loaded (/usr/lib/systemd/system/firewalld.service; disabled; vendor preset: enabled)
   Active: inactive (dead)
     Docs: man:firewalld(1)
[root@localhost ~]# getenforce 
Disabled
[root@elk2 ~]# cat /etc/hosts 
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.0.102 elk1
192.168.0.103 elk2
192.168.0.104 client
[root@localhost ~]# hostnamectl set-hostname elk2
[root@localhost ~]# bash
[root@elk2 ~]# 
[root@elk2 ~]# yum -y install java
[root@elk2 ~]# java -version
openjdk version "1.8.0_312"
OpenJDK Runtime Environment (build 1.8.0_312-b07)
OpenJDK 64-Bit Server VM (build 25.312-b07, mixed mode)

[root@localhost ~]# systemctl status firewalld
● firewalld.service - firewalld - dynamic firewall daemon
   Loaded: loaded (/usr/lib/systemd/system/firewalld.service; disabled; vendor preset: enabled)
   Active: inactive (dead)
     Docs: man:firewalld(1)
[root@localhost ~]# getenforce 
Disabled
[root@client ~]# cat /etc/hosts 
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.0.102 elk1
192.168.0.103 elk2
192.168.0.104 client
[root@localhost ~]# hostnamectl set-hostname client
[root@localhost ~]# bash
[root@client ~]# 
[root@client ~]# yum -y install java
[root@client ~]# java -version
openjdk version "1.8.0_312"
OpenJDK Runtime Environment (build 1.8.0_312-b07)
OpenJDK 64-Bit Server VM (build 25.312-b07, mixed mode)

//部署elasticsearch 下面为需要下载的软件包地址
https://phantomjs.org/download.html
https://nodejs.org/dist/v12.18.1/
https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.6.0-x86_64.rpm
https://codeload.github.com/mobz/elasticsearch-head/zip/refs/heads/master
[root@elk1 ~]# ls /opt/
elasticsearch-7.6.0-x86_64.rpm  elasticsearch-head-master.zip  node-v12.18.1-linux-x64.tar.gz  phantomjs-2.1.1-linux-x86_64.tar.bz2
[root@elk1 opt]# rpm -ivh elasticsearch-7.6.0-x86_64.rpm 
warning: elasticsearch-7.6.0-x86_64.rpm: Header V4 RSA/SHA512 Signature, key ID d88e42b4: NOKEY
Verifying...                          ################################# [100%]
Preparing...                          ################################# [100%]
Creating elasticsearch group... OK
Creating elasticsearch user... OK
Updating / installing...
   1:elasticsearch-0:7.6.0-1          ################################# [100%]
### NOT starting on installation, please execute the following statements to configure elasticsearch service to start automatically using systemd
 sudo systemctl daemon-reload
 sudo systemctl enable elasticsearch.service
### You can start elasticsearch service by executing
 sudo systemctl start elasticsearch.service
Created elasticsearch keystore in /etc/elasticsearch
[/usr/lib/tmpfiles.d/elasticsearch.conf:1] Line references path below legacy directory /var/run/, updating /var/run/elasticsearch → /run/elasticsearch; please update the tmpfiles.d/ drop-in file accordingly.
[root@elk1 opt]# cd /etc/elasticsearch/
[root@elk1 elasticsearch]# 
[root@elk1 elasticsearch]# cp elasticsearch.yml elasticsearch.yml.bak
[root@elk1 elasticsearch]# vi elasticsearch.yml
[root@elk1 elasticsearch]# 
[root@elk1 elasticsearch]# grep -v "^#" /etc/elasticsearch/elasticsearch.yml
cluster.name: sjy      #集群名称，各节点配成相同的集群名称。
node.name: elk1        #节点名称，各节点配置不同
path.data: /data/elk_data   # 数据存储目录。
path.logs: /var/log/elasticsearch   #日志存储目录。
bootstrap.memory_lock: false    #内存锁定，是否禁用交换。
network.host: 0.0.0.0    #绑定节点IP。
http.port: 9200  # rest api端口。
discovery.zen.ping.unicast.hosts: ["elk1", "elk2"]   #提供其他 Elasticsearch 服务节点的单点广播发现功能。
cluster.initial_master_nodes: ["elk1"]   #初始主节点
[root@elk1 elasticsearch]# mkdir -p /data/elk_data
[root@elk1 elasticsearch]# chown elasticsearch:elasticsearch /data/elk_data/
[root@elk1 opt]# systemctl daemon-reload
[root@elk1 elasticsearch]# systemctl start elasticsearch.service
[root@elk1 elasticsearch]# systemctl status elasticsearch.service
● elasticsearch.service - Elasticsearch
   Loaded: loaded (/usr/lib/systemd/system/elasticsearch.service; disabled; vendor preset: disabled)
   Active: active (running) since Wed 2023-07-19 10:06:19 CST; 49s ago
     Docs: http://www.elastic.co
 Main PID: 12546 (java)
    Tasks: 50 (limit: 23009)
   Memory: 1.1G
   CGroup: /system.slice/elasticsearch.service
           ├─12546 /usr/share/elasticsearch/jdk/bin/java -Des.networkaddress.cache.ttl=60 -Des.networkaddress.cache.negative.ttl=10 -XX:+AlwaysPreTouch -Xss1m -Djava.awt.headless=true -Dfile.enco>
           └─12640 /usr/share/elasticsearch/modules/x-pack-ml/platform/linux-x86_64/bin/controller
[root@elk1 elasticsearch]# systemctl enable --now elasticsearch.service
Synchronizing state of elasticsearch.service with SysV service script with /usr/lib/systemd/systemd-sysv-install.
Executing: /usr/lib/systemd/systemd-sysv-install enable elasticsearch
Created symlink /etc/systemd/system/multi-user.target.wants/elasticsearch.service → /usr/lib/systemd/system/elasticsearch.service.
[root@elk1 elasticsearch]# ss -antl
State                  Recv-Q                 Send-Q                                 Local Address:Port                                   Peer Address:Port                 Process                 
LISTEN                 0                      128                                          0.0.0.0:22                                          0.0.0.0:*                                            
LISTEN                 0                      2048                                               *:9200                                              *:*                                            
LISTEN                 0                      2048                                               *:9300                                              *:*                                            
LISTEN                 0                      128                                             [::]:22                                             [::]:*                                       


[root@elk2 ~]# ls /opt/
elasticsearch-7.6.0-x86_64.rpm  elasticsearch-head-master.zip  node-v12.18.1-linux-x64.tar.gz  phantomjs-2.1.1-linux-x86_64.tar.bz2
[root@elk2 opt]# rpm -ivh elasticsearch-7.6.0-x86_64.rpm
warning: elasticsearch-7.6.0-x86_64.rpm: Header V4 RSA/SHA512 Signature, key ID d88e42b4: NOKEY
Verifying...                          ################################# [100%]
Preparing...                          ################################# [100%]
Creating elasticsearch group... OK
Creating elasticsearch user... OK
Updating / installing...
   1:elasticsearch-0:7.6.0-1          ################################# [100%]
### NOT starting on installation, please execute the following statements to configure elasticsearch service to start automatically using systemd
 sudo systemctl daemon-reload
 sudo systemctl enable elasticsearch.service
### You can start elasticsearch service by executing
 sudo systemctl start elasticsearch.service
Created elasticsearch keystore in /etc/elasticsearch
[/usr/lib/tmpfiles.d/elasticsearch.conf:1] Line references path below legacy directory /var/run/, updating /var/run/elasticsearch → /run/elasticsearch; please update the tmpfiles.d/ drop-in file accordingly.
[root@elk2 opt]# cd /etc/elasticsearch/
[root@elk2 elasticsearch]# 
[root@elk2 elasticsearch]# cp elasticsearch.yml elasticsearch.yml.bak
[root@elk2 elasticsearch]# vi elasticsearch.yml
[root@elk2 elasticsearch]# grep -v "^#" /etc/elasticsearch/elasticsearch.yml
cluster.name: sjy
node.name: elk2
path.data: /data/elk_data
path.logs: /var/log/elasticsearch
bootstrap.memory_lock: false
network.host: 0.0.0.0
http.port: 9200
discovery.zen.ping.unicast.hosts: ["elk1", "elk2"]
cluster.initial_master_nodes: ["elk1"]
[root@elk2 elasticsearch]# mkdir -p /data/elk_data
[root@elk2 elasticsearch]# chown elasticsearch:elasticsearch /data/elk_data/
[root@elk2 opt]# systemctl daemon-reload
[root@elk2 elasticsearch]# systemctl start elasticsearch.service
[root@elk2 elasticsearch]# systemctl status elasticsearch.service
● elasticsearch.service - Elasticsearch
   Loaded: loaded (/usr/lib/systemd/system/elasticsearch.service; disabled; vendor preset: disabled)
   Active: active (running) since Wed 2023-07-19 10:13:27 CST; 7s ago
     Docs: http://www.elastic.co
 Main PID: 11400 (java)
    Tasks: 48 (limit: 23009)
   Memory: 1.2G
   CGroup: /system.slice/elasticsearch.service
           ├─11400 /usr/share/elasticsearch/jdk/bin/java -Des.networkaddress.cache.ttl=60 -Des.networkaddress.cache.negative.ttl=10 -XX:+AlwaysPreTouch -Xss1m -Djava.awt.headless=true -Dfile.enco>
           └─11495 /usr/share/elasticsearch/modules/x-pack-ml/platform/linux-x86_64/bin/controller

Jul 19 10:13:08 elk2 systemd[1]: Starting Elasticsearch...
Jul 19 10:13:09 elk2 elasticsearch[11400]: OpenJDK 64-Bit Server VM warning: Option UseConcMarkSweepGC was deprecated in version 9.0 and will likely be removed in a future release.
Jul 19 10:13:27 elk2 systemd[1]: Started Elasticsearch.
[root@elk2 elasticsearch]# systemctl enable --now elasticsearch.service
Synchronizing state of elasticsearch.service with SysV service script with /usr/lib/systemd/systemd-sysv-install.
Executing: /usr/lib/systemd/systemd-sysv-install enable elasticsearch
Created symlink /etc/systemd/system/multi-user.target.wants/elasticsearch.service → /usr/lib/systemd/system/elasticsearch.service.
[root@elk2 elasticsearch]# ss -antl
State                  Recv-Q                 Send-Q                                 Local Address:Port                                   Peer Address:Port                 Process                 
LISTEN                 0                      128                                          0.0.0.0:22                                          0.0.0.0:*                                            
LISTEN                 0                      128                                             [::]:22                                             [::]:*                                            
LISTEN                 0                      2048                                               *:9200                                              *:*                                            
LISTEN                 0                      2048                                               *:9300                                              *:*

//访问测试
部署ELK+Kafka+Filebeat日志收集分析系统,elk,kafka,分布式

//安装 Elasticsearch-head 插件 安装 node  phantomjs 安装 Elasticsearch-head 数据可视化工具 在elk1上操作
现在nodev12.8.1需要编译安装
在解压目录下执行./configure --prefix=/usr/local/node 后执行make && make install 
下面添加环境变量配置一致
[root@elk1 opt]# ls
elasticsearch-7.6.0-x86_64.rpm  elasticsearch-head-master.zip  node-v12.18.1-linux-x64.tar.gz  phantomjs-2.1.1-linux-x86_64.tar.bz2
[root@elk1 opt]# yum -y install unzip bzip2
[root@elk1 opt]# tar xf node-v12.18.1-linux-x64.tar.gz 
[root@elk1 opt]# mv node-v12.18.1-linux-x64 /usr/local/node
[root@elk1 opt]# echo "export PATH=$PATH:/usr/local/node/bin" >> ~/.bashrc
[root@elk1 opt]# source ~/.bashrc
[root@elk1 opt]# node -v
v12.18.1
[root@elk1 opt]# tar jxvf phantomjs-2.1.1-linux-x86_64.tar.bz2 -C /usr/local/src/
[root@elk1 opt]# cd /usr/local/src/phantomjs-2.1.1-linux-x86_64/bin/
[root@elk1 bin]# cp phantomjs /usr/local/bin/
[root@elk1 opt]# unzip elasticsearch-head-master.zip
//修改为淘宝的源
[root@elk1 opt]# npm config set registry http://registry.npm.taobao.org
[root@elk1 opt]# cd elasticsearch-head-master
[root@elk1 elasticsearch-head-master]# npm install
Failed at the phantomjs-prebuilt@2.1.16 install script.
npm ERR! This is probably not a problem with npm. There is likely additional logging output above.

npm ERR! A complete log of this run can be found in:
npm ERR!     /root/.npm/_logs/2023-07-19T04_16_16_954Z-debug.log
//看到这个代表成功

//修改 Elasticsearch 主配置文件
[root@elk1 elasticsearch-head-master]# vi /etc/elasticsearch/elasticsearch.yml

[root@elk1 elasticsearch-head-master]# vi Gruntfile.js 
 connect: {
                        server: {
                                options: {
                                        hostname: '192.168.0.102', //添加这行
                                        port: 9100,
                                        base: '.',
                                        keepalive: true
                                }
                        }
                }
[root@elk1 elasticsearch-head-master]# cd _site/
[root@elk1 _site]# vi app.js 
init: function(parent) {
                        this._super();
                        this.prefs = services.Preferences.instance();
                        this.base_uri = this.config.base_uri || this.prefs.get("app-base_uri") || "http://192.168.0.102:9200";   //修改这里
                        if( this.base_uri.charAt( this.base_uri.length - 1 ) !== "/" ) {
                                // XHR request fails if the URL is not ending with a "/"
                                this.base_uri += "/";
                        }
在每个节点的/etc/elasticsearch/elasticsearch.yml/配置文件最后加上
http.cors.enabled: true   //这两行 #是否允许跨源 REST 请求，用于允许head插件访问ES。
http.cors.allow-origin: "*" #允许的源地址。
//启动
[root@elk1 bin]# systemctl restart elasticsearch
[root@elk1 bin]# /root/elasticsearch-head-master/node_modules/grunt/bin
[root@elk1 bin]# ./grunt server
>> Local Npm module "grunt-contrib-jasmine" not found. Is it installed?

Running "connect:server" (connect) task
Waiting forever...
Started connect web server on http://192.168.0.102:9100
//另起一个终端查看 也可使用nohup ./grunt server & 让他在后台运行
[root@elk1 ~]# ss -antl
State                  Recv-Q                 Send-Q                                 Local Address:Port                                   Peer Address:Port                 Process                 
LISTEN                 0                      128                                          0.0.0.0:22                                          0.0.0.0:*                                            
LISTEN                 0                      511                                    192.168.0.102:9100                                        0.0.0.0:*                                            
LISTEN                 0                      2048                                               *:9200                                              *:*                                            
LISTEN                 0                      2048                                               *:9300                                              *:*                                            
LISTEN                 0                      128                                             [::]:22                                             [::]:*

部署ELK+Kafka+Filebeat日志收集分析系统,elk,kafka,分布式

//部署kabana
https://artifacts.elastic.co/downloads/kibana/kibana-7.6.0-x86_64.rpm
[root@elk1 opt]# ls
elasticsearch-7.6.0-x86_64.rpm  elasticsearch-head-master  elasticsearch-head-master.zip  kibana-7.6.0-x86_64.rpm  node-v12.18.1-linux-x64.tar.gz  phantomjs-2.1.1-linux-x86_64.tar.bz2
[root@elk1 opt]# rpm -ivh kibana-7.6.0-x86_64.rpm
[root@elk1 opt]# vi /etc/kibana/kibana.yml  //修改以下内容
server.port: 5601 #kibana服务端口，默认5601
server.host: "192.168.0.102"  #kibana主机IP地址，默认localhost
server.name: "elk-application" #kibana访问域名
elasticsearch.hosts: ["http://192.168.0.102:9200"] #es地址
i18n.locale: "zh-CN" #页面使用中文
[root@elk1 opt]# systemctl daemon-reload
[root@elk1 opt]# systemctl restart kibana
[root@elk1 opt]# systemctl enable kibana
[root@elk1 opt]# systemctl status kibana
● kibana.service - Kibana
   Loaded: loaded (/etc/systemd/system/kibana.service; enabled; vendor preset: disabled)
   Active: active (running) since Wed 2023-07-19 15:16:32 CST; 1min 30s ago
 Main PID: 13519 (node)
    Tasks: 11 (limit: 23009)
   Memory: 440.2M
   CGroup: /system.slice/kibana.service
           └─13519 /usr/share/kibana/bin/../node/bin/node /usr/share/kibana/bin/../src/cli -c /etc/kibana/kibana.yml

Jul 19 15:16:32 elk1 systemd[1]: Started Kibana.
Jul 19 15:17:24 elk1 kibana[13519]: {"type":"log","@timestamp":"2023-07-19T07:17:24Z","tags":["info","plugins-service"],"pid":13519,"message":"Plugin \"case\" is disabled."}
[root@elk1 opt]# ss -antl
State                  Recv-Q                 Send-Q                                 Local Address:Port                                   Peer Address:Port                 Process                 
LISTEN                 0                      128                                          0.0.0.0:22                                          0.0.0.0:*                                            
LISTEN                 0                      511                                    192.168.0.102:5601                                        0.0.0.0:*                                            
LISTEN                 0                      511                                    192.168.0.102:9100                                        0.0.0.0:*                                            
LISTEN                 0                      2048                                               *:9200                                              *:*                                            
LISTEN                 0                      2048                                               *:9300                                              *:*                                            
LISTEN                 0                      128                                             [::]:22                                             [::]:*

测试
部署ELK+Kafka+Filebeat日志收集分析系统,elk,kafka,分布式

//部署zookeeper
http://archive.apache.org/dist/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz
[root@elk1 opt]# ls
elasticsearch-7.6.0-x86_64.rpm  elasticsearch-head-master.zip  node-v12.18.1-linux-x64.tar.gz        zookeeper-3.4.14.tar.gz
elasticsearch-head-master       kibana-7.6.0-x86_64.rpm        phantomjs-2.1.1-linux-x86_64.tar.bz2
[root@elk1 opt]# tar xf zookeeper-3.4.14.tar.gz -C /data/
[root@elk1 opt]# mv /data/zookeeper-3.4.14 /data/zookeeper
[root@elk1 opt]# mkdir /data/zookeeper/{data,logs}
[root@elk1 opt]# ls /data/zookeeper/
bin        data             ivy.xml      logs        README.md             zookeeper-3.4.14.jar      zookeeper-3.4.14.jar.sha1  zookeeper-docs  zookeeper-recipes
build.xml  dist-maven       lib          NOTICE.txt  README_packaging.txt  zookeeper-3.4.14.jar.asc  zookeeper-client           zookeeper-it    zookeeper-server
conf       ivysettings.xml  LICENSE.txt  pom.xml     src                   zookeeper-3.4.14.jar.md5  zookeeper-contrib          zookeeper-jute
[root@elk1 opt]# cd /data/zookeeper/conf/
[root@elk1 conf]# cp zoo_sample.cfg zoo.cfg 
[root@elk1 conf]# vi zoo.cfg 
[root@elk1 conf]# grep -v "^#" zoo.cfg 
tickTime=2000 #ZK服务器之间或客户端与服务器之间维持心跳的时间间隔。
initLimit=10  #允许follower(相对于Leaderer言的“客户端”)连接并同步到Leader的初始化连接时间，以tickTime为单位。当初始化连接时间超过该值，则表示连接失败。
syncLimit=5  #Leader与Follower之间发送消息时，请求和应答时间长度。如果follower在设置时间内不能与leader通信，那么此follower将会被丢弃
dataDir=/data/zookeeper/data  #ZK数据存放目录。
DataLogDir=/data/zookeeper/logs  #ZK日志存放目录。
clientPort=2181 #客户端连接ZK服务的端口。
server.1=192.168.0.102:2888:3888 #2888是follower与leader交换信息的端口，3888是当leader挂了时用来执行选举时服务器相互通信的端口。
server.2=192.168.0.103:2888:3888
server.3=192.168.0.104:2888:3888
[root@elk1 conf]# echo 1 > /data/zookeeper/data/myid
//其余两个节点配置文件一样 id文件不同 操作也一样
[root@elk2 opt]# tar xf zookeeper-3.4.14.tar.gz -C /data/
[root@elk2 opt]# mv /data/zookeeper-3.4.14 /data/zookeeper
[root@elk2 opt]# mkdir /data/zookeeper/{data,logs}
[root@elk2 opt]# grep -v "^#" /data/zookeeper/conf/zoo.cfg 
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper/data
DataLogDir=/data/zookeeper/logs
clientPort=2181
server.1=192.168.0.102:2888:3888
server.2=192.168.0.103:2888:3888
server.3=192.168.0.104:2888:3888
[root@elk2 opt]# echo 2 > /data/zookeeper/data/myid
[root@client opt]# mkdir /data
[root@client opt]# tar xf zookeeper-3.4.14.tar.gz -C /data/
[root@client opt]# mv /data/zookeeper-3.4.14 /data/zookeeper
[root@client opt]# mkdir /data/zookeeper/{data,logs}
[root@client opt]# grep -v "^#" /data/zookeeper/conf/zoo.cfg 
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper/data
DataLogDir=/data/zookeeper/logs
clientPort=2181
server.1=192.168.0.102:2888:3888
server.2=192.168.0.103:2888:3888
server.3=192.168.0.104:2888:3888
[root@client opt]# echo 3 > /data/zookeeper/data/myid
//启动所有节点
zookeeper集群必须保证有两个节点存活，也就是必须同时启动两个节点，否则集群将启动不成功，因此要修改好配置文件后，在统一启动
[root@elk1 zookeeper]# cd bin/
[root@elk1 bin]# ls
README.txt  zkCleanup.sh  zkCli.cmd  zkCli.sh  zkEnv.cmd  zkEnv.sh  zkServer.cmd  zkServer.sh  zkTxnLogToolkit.cmd  zkTxnLogToolkit.sh
[root@elk1 bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[root@elk1 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Mode: follower

[root@elk2 opt]# cd /data/zookeeper/bin/
[root@elk2 bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[root@elk2 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Mode: follower

[root@client opt]# cd /data/zookeeper/bin/
[root@client bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[root@client bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Mode: leader

//部署kafka
https://archive.apache.org/dist/kafka/2.0.0/kafka_2.11-2.0.0.tgz
[root@elk1 opt]# tar xf kafka_2.11-2.0.0.tgz -C /data/
[root@elk1 opt]# mv /data/kafka_2.11-2.0.0 /data/kafka
[root@elk1 opt]# cd /data/kafka/
[root@elk1 kafka]# grep -v "#" config/server.properties 
broker.id=1
listeners=PLAINTEXT://192.168.0.102:9092
host.name=192.168.0.102
advertised.listeners=PLAINTEXT://192.168.0.102:9092
advertised.host.name=192.168.0.102
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/data
num.partitions=3
delete.topic.enable=true
auto.create.topics.enable=true 
replica.fetch.max.bytes=5242880
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
message.max.byte=5242880
log.cleaner.enable=true
log.retention.hours=48
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.0.102:2181,192.168.0.103:2181,192.168.0.104:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0

broker.id   每个server需要单独配置broker id，如果不配置系统会自动配置。
listeners       监听地址，格式PLAINTEXT://IP:端口。
num.network.threads 接收和发送网络信息的线程数。
num.io.threads          服务器用于处理请求的线程数，其中可能包括磁盘I/O。
socket.send.buffer.bytes    套接字服务器使用的发送缓冲区(SO_SNDBUF)
socket.receive.buffer.bytes 套接字服务器使用的接收缓冲区(SO_RCVBUF)
socket.request.max.bytes        套接字服务器将接受的请求的最大大小(防止OOM)
log.dirs        日志文件目录。
num.partitions  partition数量。
num.recovery.threads.per.data.dir       在启动时恢复日志、关闭时刷盘日志每个数据目录的线程的数量，默认1。
offsets.topic.replication.factor        偏移量话题的复制因子（设置更高保证可用），为了保证有效的复制，偏移话题的复制因子是可配置的，在偏移话题的第一次请求的时候可用的broker的数量至少为复制因子的大小，否则要么话题创建失败，要么复制因子取可用broker的数量和配置复制因子的最小值。
log.retention.hours 日志文件删除之前保留的时间（单位小时），默认168
log.segment.bytes   单个日志文件的大小，默认1073741824
log.retention.check.interval.ms 检查日志段以查看是否可以根据保留策略删除它们的时间间隔。
zookeeper.connect   ZK主机地址，如果zookeeper是集群则以逗号隔开。
zookeeper.connection.timeout.ms     连接到Zookeeper的超时时间。

[root@elk1 kafka]# mkdir /data/kafka/data

[root@elk2 opt]# tar xf kafka_2.11-2.0.0.tgz -C /data/
[root@elk2 opt]# mv /data/kafka_2.11-2.0.0 /data/kafka
[root@elk2 opt]# vi /data/kafka/config/server.properties 
broker.id=2
listeners=PLAINTEXT://192.168.0.103:9092
host.name=192.168.0.103
advertised.listeners=PLAINTEXT://192.168.0.103:9092
advertised.host.name=192.168.0.103
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/data
num.partitions=3
delete.topic.enable=true
auto.create.topics.enable=true 
replica.fetch.max.bytes=5242880
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
message.max.byte=5242880
log.cleaner.enable=true
log.retention.hours=48
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.0.102:2181,192.168.0.103:2181,192.168.0.104:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0
[root@elk2 opt]# mkdir /data/kafka/data

[root@client opt]# tar xf kafka_2.11-2.0.0.tgz -C /data/
[root@client opt]# mv /data/kafka_2.11-2.0.0 /data/kafka
[root@client opt]# vi /data/kafka/config/server.properties 
broker.id=3
listeners=PLAINTEXT://192.168.0.104:9092
host.name=192.168.0.104
advertised.listeners=PLAINTEXT://192.168.0.104:9092
advertised.host.name=192.168.0.104
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/data
num.partitions=3
delete.topic.enable=true
auto.create.topics.enable=true 
replica.fetch.max.bytes=5242880
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
message.max.byte=5242880
log.cleaner.enable=true
log.retention.hours=48
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.0.102:2181,192.168.0.103:2181,192.168.0.104:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0
[root@client opt]# mkdir /data/kafka/data
//启动kafka
[root@elk1 kafka]# cd /data/kafka/bin/
[root@elk1 bin]# ls
connect-distributed.sh        kafka-console-producer.sh    kafka-log-dirs.sh                    kafka-run-class.sh                  kafka-verifiable-producer.sh     zookeeper-shell.sh
connect-standalone.sh         kafka-consumer-groups.sh     kafka-mirror-maker.sh                kafka-server-start.sh               trogdor.sh
kafka-acls.sh                 kafka-consumer-perf-test.sh  kafka-preferred-replica-election.sh  kafka-server-stop.sh                windows
kafka-broker-api-versions.sh  kafka-delegation-tokens.sh   kafka-producer-perf-test.sh          kafka-streams-application-reset.sh  zookeeper-security-migration.sh
kafka-configs.sh              kafka-delete-records.sh      kafka-reassign-partitions.sh         kafka-topics.sh                     zookeeper-server-start.sh
kafka-console-consumer.sh     kafka-dump-log.sh            kafka-replica-verification.sh        kafka-verifiable-consumer.sh        zookeeper-server-stop.sh
[root@elk1 bin]# ./kafka-server-start.sh -daemon /data/kafka/config/server.properties
[root@elk1 bin]# ss -antl
State                 Recv-Q                Send-Q                                        Local Address:Port                                Peer Address:Port               Process                
LISTEN                0                     128                                                 0.0.0.0:22                                       0.0.0.0:*                                         
LISTEN                0                     511                                           192.168.0.102:5601                                     0.0.0.0:*                                         
LISTEN                0                     511                                           192.168.0.102:9100                                     0.0.0.0:*                                         
LISTEN                0                     50                                   [::ffff:192.168.0.102]:3888                                           *:*                                         
LISTEN                0                     2048                                                      *:9200                                           *:*                                         
LISTEN                0                     2048                                                      *:9300                                           *:*                                         
LISTEN                0                     128                                                    [::]:22                                          [::]:*                                         
LISTEN                0                     50                                                        *:39769                                          *:*                                         
LISTEN                0                     50                                                        *:33531                                          *:*                                         
LISTEN                0                     50                                   [::ffff:192.168.0.102]:9092                                           *:*                                         
LISTEN                0                     50                                                        *:2181                                           *:*                           

[root@elk2 opt]# cd /data/kafka/bin/
[root@elk2 bin]# ./kafka-server-start.sh -daemon /data/kafka/config/server.properties
[root@elk2 bin]# ss -antl
State                 Recv-Q                Send-Q                                        Local Address:Port                                Peer Address:Port               Process                
LISTEN                0                     128                                                 0.0.0.0:22                                       0.0.0.0:*                                         
LISTEN                0                     128                                                    [::]:22                                          [::]:*                                         
LISTEN                0                     50                                                        *:33723                                          *:*                                         
LISTEN                0                     50                                                        *:2181                                           *:*                                         
LISTEN                0                     50                                   [::ffff:192.168.0.103]:3888                                           *:*                                         
LISTEN                0                     2048                                                      *:9200                                           *:*                                         
LISTEN                0                     2048                                                      *:9300                                           *:*                                         
LISTEN                0                     50                                                        *:44597                                          *:*                                                    

[root@client opt]# cd /data/kafka/bin/
[root@client bin]# ./kafka-server-start.sh -daemon /data/kafka/config/server.properties
[root@client bin]# ss -antl
State                 Recv-Q                Send-Q                                        Local Address:Port                                Peer Address:Port               Process                
LISTEN                0                     128                                                 0.0.0.0:22                                       0.0.0.0:*                                         
LISTEN                0                     50                                   [::ffff:192.168.0.104]:3888                                           *:*                                         
LISTEN                0                     128                                                    [::]:22                                          [::]:*                                         
LISTEN                0                     50                                                        *:43359                                          *:*                                         
LISTEN                0                     50                                                        *:38111                                          *:*                                         
LISTEN                0                     50                                                        *:2181                                           *:*                                         
LISTEN                0                     50                                   [::ffff:192.168.0.104]:2888                                           *:*

//测试kafka与zookeeper连接
//创建一个topic
[root@elk1 bin]# ./kafka-topics.sh --create --zookeeper 192.168.0.102:2181,192.168.0.103:2181,192.168.0.104:2181 --replication-factor 1 --partitions 1 --topic testpic
Created topic "testpic".
//查看topic
[root@elk1 bin]#  ./kafka-topics.sh --list --zookeeper 192.168.0.104:2181,192.168.0.103:2181,192.168.0.104:2181
testpic
//查看topic的描述信息
[root@elk1 bin]#  ./kafka-topics.sh --describe --zookeeper 192.168.0.102:2181,192.168.0.103:2181,192.168.0.104:2181 --topic testpic
Topic:testpic	PartitionCount:1	ReplicationFactor:1	Configs:
	Topic: testpic	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
//使用kafka-console-producer控制台生产数据
[root@elk1 bin]# ./kafka-console-producer.sh --broker-list 192.168.0.102:9092,192.168.0.103:9092,192.168.0.104:9092 --topic testpic
>test1
>yes
>test2
>test3
>test4
>test5
>test6
>test7
>test8
>test9

//使用kafka-console-consumer控制台消费数据
[root@elk1 bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.0.102:9092,192.168.0.103:9092,192.168.0.104:9092 --topic testpic --from-beginning
test1
yes
test2
test3
test4
test5
test6
test7
test8
test9

//删除一个topic
[root@elk1 bin]# ./kafka-topics.sh --delete --zookeeper 192.168.0.102:2181  --topic testpic
Topic testpic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

//配置filebeat收集nginx、tomcat日志存储到kafka中 安装配置nginx
[root@client ~]# yum -y install nginx
[root@client ~]# vi /etc/nginx/nginx.conf
http {
    log_format  main '{"时间":"$time_iso8601",'
                '"客户端外网地址":"$http_x_forwarded_for",'
                '"客户端内网地址":"$remote_addr",'
                '"状态码":$status,'
                '"传输流量":$body_bytes_sent,'
                '"跳转来源":"$http_referer",'
                '"URL":"$request",'
                '"浏览器":"$http_user_agent",'
                '"请求响应时间":$request_time,'
                '"后端地址":"$upstream_addr"}';

    access_log  /var/log/nginx/access.log  main;
[root@client ~]# systemctl start nginx
[root@client ~]# systemctl enable nginx
Created symlink /etc/systemd/system/multi-user.target.wants/nginx.service → /usr/lib/systemd/system/nginx.service.
[root@client ~]# curl 127.0.0.1
[root@client ~]# tail /var/log/nginx/access.log 
{"时间":"2023-07-19T17:53:02+08:00","客户端外网地址":"-","客户端内网地址":"127.0.0.1","状态码":200,"传输流量":4057,"跳转来源":"-","URL":"GET / HTTP/1.1","浏览器":"curl/7.61.1","请求响应时间":0.000,"后端地址":"-"}

//安装tomcat
https://downloads.apache.org/tomcat/tomcat-8/v8.5.90/bin/apache-tomcat-8.5.90.tar.gz
[root@client opt]# tar xf apache-tomcat-8.5.90.tar.gz -C /data/
[root@client opt]# mv /data/apache-tomcat-8.5.90 /data/tomcat
[root@client opt]# /data/tomcat/bin/startup.sh 
Using CATALINA_BASE:   /data/tomcat
Using CATALINA_HOME:   /data/tomcat
Using CATALINA_TMPDIR: /data/tomcat/temp
Using JRE_HOME:        /usr
Using CLASSPATH:       /data/tomcat/bin/bootstrap.jar:/data/tomcat/bin/tomcat-juli.jar
Using CATALINA_OPTS:   
Tomcat started.

//安装filebeat
https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.6.0-x86_64.rpm
https://artifacts.elastic.co/downloads/logstash/logstash-7.6.0.rpm
[root@client opt]# rpm -ivh filebeat-7.6.0-x86_64.rpm 
warning: filebeat-7.6.0-x86_64.rpm: Header V4 RSA/SHA512 Signature, key ID d88e42b4: NOKEY
Verifying...                          ################################# [100%]
Preparing...                          ################################# [100%]
Updating / installing...
   1:filebeat-7.6.0-1                 ################################# [100%]
[root@client opt]# vi /etc/filebeat/filebeat.yml 
filebeat.inputs:
- type: log                                      #类型为log
  enabled: true
  paths:                                        #指定日志所在的路径
    - /var/log/nginx/access.log
  json.keys_under_root: true                    #支持json格式的日志输出
  json.overwriite_keys: true
  fields:                                       #在日志中增加一个字段，字段为log_topic，值为nginx_access，logstash根据带有这个字段的日志存储到指定的es索引库
    log_topic: nginx-access
  tail_files: true                              #开启日志监控，从日志的最后一行开始收集

- type: log
  enabled: true
  paths:
    - /data/tomcat/logs/catalina.out
  multiline.pattern: '^20'                      #收集tomcat错误日志，从第一个20到下一个20之间的日志整合在一行中显示
  multiline.negate: true
  multiline.match: after
  fields:
    log_topic: tomcat-cata
  tail_files: true

output.kafka:                                   #输出到kafka系统
  enabled: true
  hosts: ["192.168.0.102:9092","192.168.0.103:9092","192.168.0.104:9092"]                           #kafka的地址
  topic: '%{[fields][log_topic]}'               #指定将日志存储到kafka集群的哪个topic中，这里的topic值是引用在inputs中定义的fields，通过这种方式可以将不同路径的日志分别存储到不同的topic中
  partition.round_robin:
    reachable_only: false
  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000
[root@client opt]# systemctl start filebeat
[root@client opt]# systemctl enable filebeat
Synchronizing state of filebeat.service with SysV service script with /usr/lib/systemd/systemd-sysv-install.
Executing: /usr/lib/systemd/systemd-sysv-install enable filebeat
Created symlink /etc/systemd/system/multi-user.target.wants/filebeat.service → /usr/lib/systemd/system/filebeat.service.
[root@client opt]# systemctl status filebeat
● filebeat.service - Filebeat sends log files to Logstash or directly to Elasticsearch.
   Loaded: loaded (/usr/lib/systemd/system/filebeat.service; enabled; vendor preset: disabled)
   Active: active (running) since Wed 2023-07-19 18:01:48 CST; 22s ago
     Docs: https://www.elastic.co/products/beats/filebeat
 Main PID: 13616 (filebeat)
    Tasks: 9 (limit: 10931)
   Memory: 13.5M
   CGroup: /system.slice/filebeat.service
           └─13616 /usr/share/filebeat/bin/filebeat -e -c /etc/filebeat/filebeat.yml -path.home /usr/share/filebeat -path.config /etc/filebeat -path.data /var/lib/filebeat -path.logs /var/log/fi>

//产生程序日志数据观察数据是否存储kafka
[root@client opt]# yum -y install httpd-tools-2.4.37-41.module_el8.5.0+977+5653bbea.x86_64 
[root@client opt]# ab -n 1000 -c 100 http://127.0.0.1/index.html
[root@client opt]# /data/tomcat/bin/shutdown.sh
[root@client opt]# /data/tomcat/bin/startup.sh
[root@elk1 bin]# ./kafka-topics.sh --list --zookeeper 192.168.0.102:2181,192.168.0.103:2181,192.168.0.104:2181
__consumer_offsets
nginx-access
tomcat-cata
//nginx-access以及tomcat-cata的topic已经创建成功

//观察kafka日志
[root@elk1 kafka]# tail -f logs/kafkaServer.out 
[2023-07-19 18:05:49,763] INFO [ReplicaAlterLogDirsManager on broker 1] Added fetcher for partitions List() (kafka.server.ReplicaAlterLogDirsManager)
[2023-07-19 18:05:52,983] INFO Updated PartitionLeaderEpoch. New: {epoch:0, offset:0}, Current: {epoch:-1, offset:-1} for Partition: nginx-access-1. Cache now contains 0 entries. (kafka.server.epoch.LeaderEpochFileCache)
[2023-07-19 18:06:09,784] INFO [ReplicaFetcherManager on broker 1] Removed fetcher for partitions tomcat-cata-2 (kafka.server.ReplicaFetcherManager)
[2023-07-19 18:06:09,958] INFO [Log partition=tomcat-cata-2, dir=/data/kafka/data] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2023-07-19 18:06:09,958] INFO [Log partition=tomcat-cata-2, dir=/data/kafka/data] Completed load of log with 1 segments, log start offset 0 and log end offset 0 in 2 ms (kafka.log.Log)
[2023-07-19 18:06:09,960] INFO Created log for partition tomcat-cata-2 in /data/kafka/data with properties {compression.type -> producer, message.format.version -> 2.0-IV1, file.delete.delay.ms -> 60000, max.message.bytes -> 1000012, min.compaction.lag.ms -> 0, message.timestamp.type -> CreateTime, message.downconversion.enable -> true, min.insync.replicas -> 1, segment.jitter.ms -> 0, preallocate -> false, min.cleanable.dirty.ratio -> 0.5, index.interval.bytes -> 4096, unclean.leader.election.enable -> false, retention.bytes -> -1, delete.retention.ms -> 86400000, cleanup.policy -> [delete], flush.ms -> 9223372036854775807, segment.ms -> 604800000, segment.bytes -> 1073741824, retention.ms -> 172800000, message.timestamp.difference.max.ms -> 9223372036854775807, segment.index.bytes -> 10485760, flush.messages -> 9223372036854775807}. (kafka.log.LogManager)
[2023-07-19 18:06:09,965] INFO [Partition tomcat-cata-2 broker=1] No checkpointed highwatermark is found for partition tomcat-cata-2 (kafka.cluster.Partition)
[2023-07-19 18:06:09,965] INFO Replica loaded for partition tomcat-cata-2 with initial high watermark 0 (kafka.cluster.Replica)
[2023-07-19 18:06:09,965] INFO [Partition tomcat-cata-2 broker=1] tomcat-cata-2 starts at Leader Epoch 0 from offset 0. Previous Leader Epoch was: -1 (kafka.cluster.Partition)
[2023-07-19 18:06:09,967] INFO [ReplicaAlterLogDirsManager on broker 1] Added fetcher for partitions List() (kafka.server.ReplicaAlterLogDirsManager)

//配置logstash从kafka中读取数据并存储到es集群
部署logstash，配置logstash从kafka中读取topic数据并存储es集群

//部署logstash
[root@client opt]# rpm -ivh logstash-7.6.0.rpm 
warning: logstash-7.6.0.rpm: Header V4 RSA/SHA512 Signature, key ID d88e42b4: NOKEY
Verifying...                          ################################# [100%]
Preparing...                          ################################# [100%]
Updating / installing...
   1:logstash-1:7.6.0-1               ################################# [100%]
Using provided startup.options file: /etc/logstash/startup.options
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/pleaserun-0.0.30/lib/pleaserun/platform/base.rb:112: warning: constant ::Fixnum is deprecated
Successfully created system startup script for Logstash
//配置logstash从kafka读取数据存储到es集群
[root@client opt]# vi /etc/logstash/conf.d/in_kafka_to_es.conf
[root@client opt]# cat /etc/logstash/conf.d/in_kafka_to_es.conf 
#从kafka中读取日志数据
input {				#数据源端
	kafka {				#类型为kafka
		bootstrap_servers => ["192.168.0.102:9092,192.168.0.103:9092,192.168.0.104:9092"]			#kafka集群地址
		topics => ["nginx-access","tomcat-cata"]			#要读取那些kafka topics
		codec => "json"										#处理json格式的数据
		auto_offset_reset => "latest"						#只消费最新的kafka数据
	}
}

#处理数据，去掉没用的字段
filter {
	if[fields][log_topic] == "nginx-access" {			#如果log_topic字段为nginx-access则进行以下数据处理
	    json {					#json格式数据处理
	         source => "message"			#source等于message的
	         remove_field => ["@version","path","beat","input","log","offset","prospector","source","tags"]			#删除指定的字段
	    }
	    mutate {			#修改数据
	         remove_field => ["_index","_id","_type","_version","_score","referer","agent"]			#删除没用的字段
	    }
	}
	
	if[fields][log_topic] == "tomcat-cata" {		#如果log_topic字段为tomcat-cata
	    grok {				#解析格式
		 match => {
		      "message" => "(?<时间>20[0-9]{2}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}) \[(?<线程名称>[^\s]{0,})\] (?<日志等级>\w+) (?<类名称>[^\s]{0,}) (?<日志详情>[\W\w]+)" 	  #将message的值增加上一些格式
		 }
	    }
	    mutate {			#修改数据
                 remove_field => ["_index","_id","_type","_version","_score","referer","agent"]			#删除没用的字段
            }  
	}
}

#数据处理后存储es集群
output {				#目标端
	if[fields][log_topic] == "nginx-access" {			#如果log_topic的字段值为nginx-access就存到下面的es集群里
	    elasticsearch {						
		action => "index"			#类型为索引
		hosts => ["192.168.0.102:9200","192.168.0.103:9200"]		#es集群地址
		index => "nginx-access-%{+YYYY.MM.dd}"			#存储到es集群的哪个索引里
		codec => "json"						#处理json格式的解析
	    } 
	}
	
	if[fields][log_topic] == "tomcat-cata" {				#如果log_topic的字段值为tomcat-cata就存到下面的es集群里
	    elasticsearch {
		action => "index"				#类型为索引
		hosts => ["192.168.0.102:9200","192.168.0.103:9200"]			#es集群地址
		index => "tomcat-cata-%{+YYYY.MM.dd}"			#存储到es集群的哪个索引里
		codec => "json"						#处理json格式的解析
	    } 
	}	    
}
[root@client logstash]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/in_kafka_to_es.conf
//启动logstash并观察日志

/etc/logstash/conf.d/in_kafka_to_es.conf
	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

[INFO ] 2023-07-20 09:15:38.121 [Agent thread] agent - Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[INFO ] 2023-07-20 09:15:38.320 [[main]<kafka] AppInfoParser - Kafka version: 2.3.0
[INFO ] 2023-07-20 09:15:38.320 [[main]<kafka] AppInfoParser - Kafka commitId: fc1aaa116b661c8a
[INFO ] 2023-07-20 09:15:38.320 [[main]<kafka] AppInfoParser - Kafka startTimeMs: 1689815738309
[INFO ] 2023-07-20 09:15:38.429 [Ruby-0-Thread-14: :1] KafkaConsumer - [Consumer clientId=logstash-0, groupId=logstash] Subscribed to topic(s): nginx-access, tomcat-cata
[INFO ] 2023-07-20 09:15:40.725 [Api Webserver] agent - Successfully started Logstash API endpoint {:port=>9600}
[INFO ] 2023-07-20 09:15:41.301 [Ruby-0-Thread-14: :1] Metadata - [Consumer clientId=logstash-0, groupId=logstash] Cluster ID: Y3evIgStTDCDwhstOK0HEw
[INFO ] 2023-07-20 09:15:41.314 [Ruby-0-Thread-14: :1] AbstractCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Discovered group coordinator 192.168.0.104:9092 (id: 2147483644 rack: null)
[INFO ] 2023-07-20 09:15:41.361 [Ruby-0-Thread-14: :1] ConsumerCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Revoking previously assigned partitions []
[INFO ] 2023-07-20 09:15:41.361 [Ruby-0-Thread-14: :1] AbstractCoordinator - [Consumer clientId=logstash-0, groupId=logstash] (Re-)joining group
[INFO ] 2023-07-20 09:15:44.785 [Ruby-0-Thread-14: :1] AbstractCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Successfully joined group with generation 1
[INFO ] 2023-07-20 09:15:44.796 [Ruby-0-Thread-14: :1] ConsumerCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Setting newly assigned partitions: tomcat-cata-0, tomcat-cata-1, tomcat-cata-2, nginx-access-0, nginx-access-1, nginx-access-2
[INFO ] 2023-07-20 09:15:45.016 [Ruby-0-Thread-14: :1] ConsumerCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Found no committed offset for partition tomcat-cata-0
[INFO ] 2023-07-20 09:15:45.016 [Ruby-0-Thread-14: :1] ConsumerCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Found no committed offset for partition tomcat-cata-1
[INFO ] 2023-07-20 09:15:45.016 [Ruby-0-Thread-14: :1] ConsumerCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Found no committed offset for partition tomcat-cata-2
[INFO ] 2023-07-20 09:15:45.016 [Ruby-0-Thread-14: :1] ConsumerCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Found no committed offset for partition nginx-access-0
[INFO ] 2023-07-20 09:15:45.016 [Ruby-0-Thread-14: :1] ConsumerCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Found no committed offset for partition nginx-access-1
[INFO ] 2023-07-20 09:15:45.016 [Ruby-0-Thread-14: :1] ConsumerCoordinator - [Consumer clientId=logstash-0, groupId=logstash] Found no committed offset for partition nginx-access-2
[INFO ] 2023-07-20 09:15:45.084 [Ruby-0-Thread-14: :1] SubscriptionState - [Consumer clientId=logstash-0, groupId=logstash] Resetting offset for partition tomcat-cata-0 to offset 1.
[INFO ] 2023-07-20 09:15:45.084 [Ruby-0-Thread-14: :1] SubscriptionState - [Consumer clientId=logstash-0, groupId=logstash] Resetting offset for partition nginx-access-2 to offset 333.
[INFO ] 2023-07-20 09:15:45.299 [Ruby-0-Thread-14: :1] SubscriptionState - [Consumer clientId=logstash-0, groupId=logstash] Resetting offset for partition tomcat-cata-2 to offset 0.
[INFO ] 2023-07-20 09:15:45.299 [Ruby-0-Thread-14: :1] SubscriptionState - [Consumer clientId=logstash-0, groupId=logstash] Resetting offset for partition nginx-access-1 to offset 333.
[INFO ] 2023-07-20 09:15:46.512 [Ruby-0-Thread-14: :1] SubscriptionState - [Consumer clientId=logstash-0, groupId=logstash] Resetting offset for partition tomcat-cata-1 to offset 0.
[INFO ] 2023-07-20 09:15:46.512 [Ruby-0-Thread-14: :1] SubscriptionState - [Consumer clientId=logstash-0, groupId=logstash] Resetting offset for partition nginx-access-0 to offset 334.

查看elasticsearch集群是否增加了对应的索引库
部署ELK+Kafka+Filebeat日志收集分析系统,elk,kafka,分布式
在kibana上关联elasticsearch索引库浏览日志数据

填写索引名
采用通配符的方式

添加一个时间筛选字段

创建成功

同样的方式创建tomcat-cata索引
部署ELK+Kafka+Filebeat日志收集分析系统,elk,kafka,分布式
查询nginx-access索引日志数据

查询tomcat-cata索引日志数据

kafka zookeeper logstash的service文件文章来源地址https://www.toymoban.com/news/detail-600593.html

[Unit]
Description=Kafka server daemon
Wants=network.target

[Service]
Type=forking
Environment="PATH=/usr/local/sbin/:/usr/local/bin:/usr/bin:/sbin:/bin:/usr/bin/java"  //根据自己安装的java配写路径
ExecStart=/data/kafka/bin/kafka-server-start.sh -daemon /data/kafka/config/server.properties
ExecStop=/data/kafka/bin/kafka-server-stop.sh
ExecReload=/bin/kill -s HUP $MATNPID
PrivateTmp=true

[Install]
WantedBy=multi-user.target


[Unit]
Description=Zookeeper server daemon
Wants=network.target

[Service]
Type=forking
Environment="PATH=/usr/local/sbin/:/usr/local/bin:/usr/bin:/sbin:/bin:/usr/bin/java"
ExecStart=/data/zookeeper/bin/zkServer.sh start
ExecStop=/data/zookeeper/bin/zkServer.sh stop
ExecReload=/data/zookeeper/bin/zkServer.sh restart

[Install]
WantedBy=multi-user.target

[Unit]
Description=Logstash server daemon
Wants=network.target

[Service]
Type=forking
Environment="PATH=/usr/local/sbin/:/usr/local/bin:/usr/bin:/sbin:/bin:/usr/bin/java"
ExecStart=/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/in_kafka_to_es.conf
ExecReload=/bin/kill -s HUP $MATNPID
PrivateTmp=true

[Install]
WantedBy=multi-user.target