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

今日内容:
    - 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