一、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
版权归原作者 Bean295 所有， 如有侵权，请联系我们删除。