一、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

推荐链接

评论可见,请评论后查看内容,谢谢!!!
 您阅读本篇文章共花了: