当前位置：首页 > news >正文

kafka配合ElasticStack技术栈的搭配使用

news 2025/8/25 0:24:24

今日内容:
   - kafka生产环境调优;
   - kafka配合ElasticStack技术栈的搭配使用;
   - zookeeper集群部署;
   - zookeeper的ACL;
   - zookeeper的调优;
   - PB级别项目;
   - ES8集群搭建/elk; (待定...)

订阅1个的topic:
   老男孩: 10

多个不同的主题分配如下: (linux82-elk: 3)
   c1: 0 3 6 9
   c2: 1 4 7
   c3: 2 5 8

订阅多个的topic:
   老男孩: 10 t1
   Linux: 10 t2
   Python: 10 t3

多个不同的主题分配如下: (linux82-elk: 3)
   c1: t1_0 ... t2_0 ... t3_0 ...
   c2: t1_1 ... t2_1 ...
   c3: t1_2 ... t2_2 ...

订阅1个的topic:
   老男孩: 10

多个不同的主题分配如下: (linux82-elk: 3)
   c1: 0 1 2
   c2: 3 4 5
   c3: 6 7 8 9

10 / 3 = 3 .. 1 ===>
999 / 100 = 9 .. 99

JBOD ---> RAID 0
---> RAID

BOND ---> ...

注意的参数:
   log.dirs
   auto.create.topics.enable
   zookeeper.connect
   num.io.threads


ElasticStack集成kafka实战案例:
   1.创建topic
kafka-topics.sh --bootstrap-server 10.0.0.103:9092 --create --topic oldboyedu-linux82-kafka-000001

2.使用filebeat收集日志到kafka集群:
cat > config/28-nginx-to-kafka.yaml <<EOF
filebeat.inputs:
- type: log
paths:
- /var/log/nginx/access.log*
json.keys_under_root: true

output.kafka:
# 写入kafka集群的地址
hosts:
- 10.0.0.102:9092
- 10.0.0.103:9092

# 写入集群的topic
topic: "oldboyedu-linux82-kafka-000001"
EOF

3.使用logstash收集kafka日志
cat > config/18-kafka-to-es.conf <<EOF
input {
kafka {
# 指定kafka的集群
bootstrap_servers => "10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092"
# 从哪个topic消费数据
topics => ["oldboyedu-linux82-kafka-000001"]
# 指定消费者组
group_id => "oldboyedu-linux82-logstash"
}
}

filter {
json {
source => "message"
remove_field => ["tags","@version","ecs","agent","input","message"]
}

geoip {
source => "clientip"
}

date {
match => [
"@oldboyedu-timestamp",
"yyyy-MM-dd'T'HH:mm:ssZ"
]
}

useragent {
source => "http_user_agent"
target => "oldboyedu-linux82-useragent"
}

}

output {
stdout {}

elasticsearch {
hosts => ["10.0.0.101:9200","10.0.0.102:9200","10.0.0.103:9200"]
index => "oldboyedu-linux82-project-kafka"
user => "elastic"
password => "123456"
}
}
EOF

zookeeper集群部署:
   1.创建zookeeper的数据目录
install -d /oldboyedu/data/zk
data_rsync.sh /oldboyedu/data/zk/

2.修改单点zk的配置文件
vim /oldboyedu/softwares/zk/conf/zoo.cfg
...
# 定义最小单元的时间范围tick。
tickTime=2000
# 启动时最长等待tick数量。
initLimit=5
# 数据同步时最长等待的tick时间进行响应ACK
syncLimit=2
# 指定数据目录
dataDir=/oldboyedu/data/zk
# 监听端口
clientPort=2181
# 开启四字命令允许所有的节点访问。
4lw.commands.whitelist=*
# server.ID=A:B:C[:D]
# ID:
# zk的唯一编号。
# A:
# zk的主机地址。
# B:
# leader的选举端口，是谁leader角色，就会监听该端口。
# C:
# 数据通信端口。
# D:
# 可选配置，指定角色。
server.101=10.0.0.101:2888:3888
server.102=10.0.0.102:2888:3888
server.103=10.0.0.103:2888:3888

   3.同步数据
data_rsync.sh /oldboyedu/softwares/apache-zookeeper-3.8.0-bin/


   4.创建myid文件
for ((host_id=101;host_id<=103;host_id++)) do ssh 10.0.0.${host_id} "echo ${host_id} > /oldboyedu/data/zk/myid";done

   5.所有节点启动zk服务
zkServer.sh start
zkServer.sh status

6.链接方式
zkCli.sh -server 10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181

zookeeper.connect=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181/oldboyedu-linux82-kafka3.2.1

zookeeper的leader选举流程:
   myid:
       唯一标识一个zookeeper节点。
   zxid:
       唯一事务的标识。用于记录唯一的写操作！

   选举流程:
       (1)相比较zxid，若zxid较大，则会成为新的leader;
       (2)如果zxid比较不出来，则比较myid，myid较大者会有限成为新的leader;


使用zkWeb管理zookeeper集群:
   1.运行zkWeb
java -jar zkWeb-v1.2.1.jar &>/dev/null &

2.访问webUI
略。

临时znode：
   当前的会话退出时，znode会默认等待30秒后自动消失，等待时间是可以修改的哟。

持久的znode:
   不会随着客户端的退出而删除znode。

docker

查看全文

http://www.lryc.cn/news/157455.html