ElasticStack日志分析平台－ES 集群、Kibana与Kafka

一、Elasticsearch

1、介绍：

Elasticsearch 是一个开源的分布式搜索和分析引擎，Logstash 和 Beats 收集的数据可以存储在 Elasticsearch 中进行搜索和分析。

Elasticsearch为所有类型的数据提供近乎实时的搜索和分析：一旦数据被索引，它就可以立即被搜索和分析，这种实时性使得用户能够即时获取最新数据的搜索结果和分析信息。

2、概念：

① 文档：文档是 Elasticsearch中所有可搜索数据的最小的数据单元。它是以JSON 格式表示的一条数据记录，每个文档都有一个唯一的ID来标识，文档可以包含各种字段，例如文本、数字、日期、嵌套对象等。

② 文档元数据：文档除了包含实际的数据外，还有一些元数据信息。这些信息包括文档的版本号、索引的时间戳、路由信息等。

③ 索引（Index）：索引是一组具有相似特征的文档的集合，每个索引都有一个唯一的名称，用于标识和检索其中的文档。

二、ES 集群：

1、概念：

Elasticsearch（ES）集群是由一个或多个Elasticsearch节点组成的集合。这些节点协同工作，共同承载数据存储、处理和搜索的负载。

(1) Master Node 和 Master-eligible Node:

① Master Node：主节点负责管理整个集群的状态和拓扑结构；

② Master-eligible Node：有资格成为主节点的节点。

ES 集群每个节点启动后，默认就是一个 Master eligible node，Master-eligible node 可以参加选主流程，成为 Master Node ；当第一个节点启动时，它会将自己选举成 Master Node，每个节点上都保存了集群的状态，只有 Master Node 才能修改集群的状态信息。

(2) Date Node 和 Coordinating Node：

① Data Node：数据节点，这些节点负责存储数据分片；

② Coordinating Node：协调节点，负责接收来自客户端的请求，将请求转发到适当的数据节点，并将数据节点的响应整合后返回给客户端。

(3) 主分片与副本分片：

分片（Shards）是将数据水平分割和分布式存储的机制，它允许将大量数据分散到集群中的多个节点上，以提高性能、扩展性和可用性。

① 主分片（Primary Shard）：主分片是索引的原始数据分片，在创建索引时，可以指定主分片的数量，一旦创建后，主分片的数量将保持不变。

② 副本分片（Replica Shard）：副本分片是主分片的复制品，用于提供数据的冗余备份和高可用性。副本分片数量可以随时更改，通过增加或减少副本的数量，可以影响集群服务的可用性。

2、集群配置：

(1) 域名解析：

(2) 节点设置（以 ela1 为例）：

vim /etc/elasticsearch/elasticsearch.yml

① cluster.name: elk 集群名称，各节点的集群名称相同

② node.name: ela1 节点名称，各节点需要设置独立的节点名称

③ node.data: true 指示节点为数据节点

④ network.host: 0.0.0.0 ；http.port: 9200 节点所绑定的ip和提供服务的端口

⑤ discovery.seed_hosts 指定集群成员，用于主动发现他们

⑥ cluster.initial_master_nodes: ["ela1", "ela2", "ela3"] 指定集群初始化主节点

2号、3号节点的 yml 文件除了 node.name 需要修改，其他都与 ela1 保持一致

(3) 查集群：

systemctl start elasticsearch.service

● 查看集群健康状态：

curl -X GET "localhost:9200/_cat/health?v"

状态含义：

绿色表示健康状态良好，黄色表示有一些问题但仍在正常运行，红色表示存在严重问题。

● 查看集群节点信息：

curl -X GET "localhost:9200/_cat/nodes?v"

3、集群测试

使用 Filebeat 搜集日志，输出到 Logstash, 再由 Logstash 处理完数据后输出到 Elasticsearch。

(1) Logstash配置：

(2) 启动服务，查看输出：

验证Elasticsearch是否创建索引：

curl ‐X GET "192.168.19.20:9200/_cat/indices"

(3) 自定义索引：

用户可自定义索引，例如把访问日志 access.log 中的内容单独放到一个索引中。

三、Kibana

1、简介：

Kibana是一个开源的数据可视化工具，它是Elastic Stack的一部分，Kibana主要用于对Elasticsearch中的数据进行可视化和分析。

2、Kibana 部署

(1) 编辑配置文件：

vim /usr/local/kibana/config/kibana.yml

① server.port: 5601 ；server.host: 0.0.0.0

② elasticsearch.hosts: ["http://192.168.198.128:9200"]

用于连接到 ES 集群的地址和端口

③ logging.dest: /var/log/kibana/kibana.log

配置日志文件路径

④ i18n.locale: "zh-CN"

设置页面字体为中文

(2) 创建用户并修改属主和属组：

默认情况下 kibana 不以 root 用户运行，需要创建应该普通用户

useradd ela

mkdir /run/kibana /var/log/kibana/

chown -R ela.ela /run/kibana/

chown -R ela.ela /var/log/kibana/

chown -R ela.ela /usr/local/kibana/

(3) 切换到 ela 用户运行：

su ‐ ela

/usr/local/kibana/bin/kibana

kibana web 界面：

① 创建索引模式：

Kibana 中创建索引模式时，你需要指定一个或多个索引模式名称，这些名称匹配 Elasticsearch中 的一个或多个索引。通过指定索引模式，可以在Kibana中执行搜索、创建可视化图表和构建仪表板。

② 查看日志：

四、Kafka 集群

1、简介：

(1) 概念：

Kafka是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它可以实时的处理大量数据以满足各种需求场景，提高了可扩展性。具有峰值处理能力，能够使关键组件顶住突发的访问压力，不会因为超负荷的请求而完全崩溃。

(2) 特性：

高吞吐量：kafka每秒可以处理几十万条消息

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

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

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

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

(3) 组件：

① 话题 (topic)：特定类型的消息流，Kafka消息被发布到话题，消费者可以订阅并从主题中读取数据（kafka 是面向 topic的）

② 生产者 (Producer) ：负责将数据发布到Kafka的话题中，生产者可以将消息发送到一个或多个话题。

③ 消费者 (Consumer)：订阅一个或多个话题，并处理从这些话题中接收到的消息。

④ 代理 (Broker)：Kafka集群由多个服务器节点组成，每个节点称为代理。它们负责存储已发布消息记录，并处理生产者和消费者之间的数据传输。

⑤ 分区 (Partition)：每个主题可以分为多个分区，每个分区在多个服务器节点上进行副本备份，确保数据的可靠性和容错性。

⑥ 复制 (Replication)：Kafka使用复制机制来确保数据的可靠性和容错性，复制允许将相同分区的数据副本保存在多个Broker上。

⑦ 领导者 (Leader)：对于每个分区，Kafka中有一个 Leader ，它负责处理所有的读写请求，所有的生产者和消费者都与 Leader 交互。

⑧ 跟随者 (Follower)：Follower 是 Leader 的复制品，Follower 会从 Leader 中拉取数据，并保持数据的同步，以便在 Leader 副本失败时接管服务。

⑨ ZooKeeper：ZooKeeper是一个开源的分布式协调服务，用于管理和协调Kafka集群中的Broker节点。ZooKeeper负责维护Kafka集群中各个Broker的状态信息，包括分区分配、Leader选举等，确保Kafka集群的稳定运行。

2、集群部署：

(1) 域名解析：

配置jdk8：yum install -y java-1.8.0-openjdk

(2) 配置 ZK（以 es01 为例）：

vim /usr/local/kafka/config/zookeeper.properties

dataDir=/opt/data/zookeeper/data

dataLogDir=/opt/data/zookeeper/logs

clientPort=2181

tickTime=2000

initLimit=20

syncLimit=10

dataDir：指定 ZooKeeper 数据存储的目录；

dataLogDir：指定 ZooKeeper 日志文件存储的目录；

clientPort：ZooKeeper 客户端连接到服务器的端口号；

tickTime：ZooKeeper 服务器之间的心跳时间以及超时时间 (ms)；

initLimit：当 ZooKeeper 服务器启动时，等待接收来自 Leader 的初始化连接的时间 (以 tickTime 的倍数为单位)；

syncLimit：在 ZooKeeper 集合中的 Follower 节点同步到 Leader 节点的时间限制 (以 tickTime 的倍数为单位)。

server.1=192.168.198.128:2888:3888

server.2=192.168.198.129:2888:3888

server.3=192.168.198.130:2888:3888

kafka集群IP:Port

● 创建data、log目录：

mkdir ‐p /opt/data/zookeeper/{data,logs}

● 创建myid文件：

指定该设备在集群中的编号

echo 1 > /opt/data/zookeeper/data/myid

es02、es03 的配置信息与 es01 相同

mkdir -p /opt/data/zookeeper/{data,logs}

echo 2 > /opt/data/zookeeper/data/myid

echo 3 > /opt/data/zookeeper/data/myid

(3) 配置 Kafka（以 es01 为例）：

vim /usr/local/kafka/config/server.properties

broker.id=1

listeners=PLAINTEXT://192.168.198.128:9092

num.network.threads=3

num.io.threads=8

broker.id：配置了 Kafka broker（代理）的唯一标识符，每个 Kafka broker 通过 broker.id 来标识自己在集群中的位置；

listeners=PLAINTEXT://192.168.198.128:9092：配置了 Kafka broker 监听客户端连接的网络接口和地址（配置本机）；

num.network.threads：这个参数设置了 Kafka broker 处理消息的最大线程数。

num.io.threads：这个参数定义了 Kafka broker 处理磁盘 I/O 操作的线程数。

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

socket.send.buffer.bytes：设置了 Kafka 发送数据时的缓存大小；

socket.receive.buffer.bytes：设置了 Kafka 接收数据时的缓存大小；

socket.request.max.bytes：设置了 Kafka 单个请求可以包含的最大字节量。

log.dirs=/opt/data/kafka/logs

num.partitions=6

num.recovery.threads.per.data.dir=1

log.dirs：Kafka 的日志存储目录配置；

num.partition：指定 Kafka 话题的分区数量；

num.recovery.threads.per.data.dir：配置每个数据目录（data.dirs）下用于日志恢复的线程数。

offsets.topic.replication.factor=2

transaction.state.log.replication.factor=1

transaction.state.log.min.isr=1

offsets.topic.replication.factor：指定了存储位置的复制方式，设置为 2 数据会被复制两份；

transaction.state.log.replication.factor ；transaction.state.log.min.isr：设置了事务状态日志数据备份数量为1份，只要有一个备份是活跃的就可以进行写入操作。

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

log.retention.hours：日志的保留时间 168小时（7天）；

log.segment.bytes：定义了日志段的大小（存储消息的字节数）；

log.retention.check.interval.ms：定义了检查日志保留策略的时间间隔，每300,000毫秒 (5分钟) 检查并清理过期的日志段。

zookeeper.connect=192.168.198.128:2181,192.168.198.129:2181,192.168.198.130:2181

zookeeper.connection.timeout.ms=6000

group.initial.rebalance.delay.ms=0

zookeeper.connect：指定了 Kafka 使用的 ZooKeeper 的连接信息；

zookeeper.connection.timeout.ms：定义了连接 ZooKeeper 的超时时间；

group.initial.rebalance.delay.ms：定义了消费者组的初始重新平衡（rebalance）的延迟时间，为 0 毫秒意味着当有新的消费者加入或退出消费者组时，Kafka 将立即开始重新平衡分配分区（partitions）给消费者。

mkdir -p /opt/data/kafka/logs

● 配置 es02 和 es03：

scp -r /usr/local/kafka/config/server.properties es02:/usr/local/kafka/config/

scp -r /usr/local/kafka/config/server.properties es03:/usr/local/kafka/config/

分别修改 broker.id 和 listeners。

mkdir -p /opt/data/kafka/logs

(4) 三台机器启动 zookeeper：

cd /usr/local/kafka

nohup bin/zookeeper‐server‐start.sh config/zookeeper.properties &

(5) 三台集群启动 kafka：

cd /usr/local/kafka

nohup bin/kafka‐server‐start.sh config/server.properties &

(6) 验证效果：

① 在1号机(128)上创建topic：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtopicBean

② 在其他机器上查看 topic：

bin/kafka-topics.sh --zookeeper 192.168.198.128:2181 --list

③ 模拟消息生产和消费 (128 发送消息，129接收)：

128：bin/kafka-console-producer.sh --broker-list 192.168.198.129:9092 --topic testtopicBean

129：bin/kafka-console-consumer.sh --bootstrap-server 192.168.198.128:9092 --topic testtopicBean --from-beginning