安装部署
先下载zookeeper、kafka
zookeeper下载地址：Download
kafka下载地址：Download
集群规划
| hadoop001 |
hadoop002 |
| zk |
zk |
| kafka |
kafka |
集群部署
1) 解压安装包
[root@hadoop001 software]# pwd
/root/software
[root@hadoop001 software]# tar -xvf kafka_2.12-2.5.0.tgz
2) 修改解压后的文件名称并移动到指定目录
[root@hadoop001 software]# mv kafka_2.12-2.5.0 kafka_2.12
[root@hadoop001 software]# mv kafka_2.12 /usr/local/
3) 在/opt/kafka目录下创建logs文件夹
[root@hadoop001 opt]# mkdir -p /opt/kafka/logs
4) 修改配置文件
#broker的全局唯一编号，不能重复
broker.id=0
#删除topic功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志存放的路径
log.dirs=/opt/kafka/logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=hadoop001:2181,hadoop002:2181
5) 配置环境变量
[root@hadoop001 kafka_2.12]# vim /etc/profile
export KAFKA_HOME=/usr/local/kafka_2.12
export PATH=$PATH:$KAFKA_HOME/bin
# 重新加载配置文件
[root@hadoop001 kafka_2.12]# source /etc/profile
6) 分发安装包
将hadoop001的kafka安装包发送到hadoop002机器上
# 首先将hadoop002对应的ip，配知道hadoop001的机器上
[root@hadoop001 local]# vim /etc/hosts
192.168.78.3 hadoop001
192.168.78.3 localhost
192.168.78.4 hadoop002
然后使用scp命令
scp -r /usr/local/kafka_2.12/ root@hadoop002:/usr/local/
分发之后记得配置其他机器的环境变量
7) 修改hadoop002的配置
到hadoop002机器上，修改broker.id=1
注：broker.id不得重复
切换到hadoop002机器上
# 增加hadoop001的IP映射
[root@192 config]# vim /etc/hosts
192.168.78.4 hadoop002
192.168.78.4 localhost
192.168.78.3 hadoop001
8) 启动集群
依次在hadoop001、hadoop001 节点上启动kafka（保证每台机器上zookeeper是启动的）
# hadoop001机器
[root@hadoop001 kafka_2.12]# bin/kafka-server-start.sh -daemon config/server.properties
# hadoop002机器
[root@hadoop002 kafka_2.12]# bin/kafka-server-start.sh -daemon config/server.properties
9) 关闭集群
[root@hadoop001 kafka_2.12]# bin/kafka-server-stop.sh
[root@hadoop002 kafka_2.12]# bin/kafka-server-stop.sh
Kafka命令行操作
1) 查看当前服务器中的所有topic
[root@hadoop001 kafka_2.12]# bin/kafka-topics.sh --zookeeper hadoop001:2181 --list
__consumer_offsets
yiyang
2) 创建topic
[root@hadoop001 kafka_2.12]# bin/kafka-topics.sh --zookeeper hadoop001:2181 --create --replication-factor 2 --partitions 1 --topic first
--topic 定义topic名
--replication-factor 定义副本的数（不能超过集群的数量）
--partition 定义分区数
3) 删除topic
[root@hadoop001 kafka_2.12]# bin/kafka-topics.sh --zookeeper hadoop001:2181 --delete --topic first
Topic first is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
需要server.properties中设置delete.topic.enable=true否则只是标记删除。
4) 发送消息
[root@hadoop001 kafka_2.12]# bin/kafka-console-producer.sh --broker-list hadoop001:9092 --topic yiyang
>hello kafka
>
5) 消费消息
[root@hadoop001 kafka_2.12]# bin/kafka-console-consumer.sh --bootstrap-server hadoop001:9092 --from-beginning --topic yiyang
hello kafka
--from-beginning：会把主题中以往所有的数据都读取出来。
6) 查看某个Topic的详情
[root@hadoop001 kafka_2.12]# bin/kafka-topics.sh -zookeeper hadoop001:2181 --describe --topic yiyang
Topic: yiyang PartitionCount: 1 ReplicationFactor: 1 Configs:
Topic: yiyang Partition: 0 Leader: 0 Replicas: 0 Isr: 0
7) 修改分区数
[root@hadoop001 kafka_2.12]# bin/kafka-topics.sh -zookeeper hadoop001:2181 --alter --topic yiyang --partitions 6
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!
[root@hadoop001 kafka_2.12]# bin/kafka-topics.sh -zookeeper hadoop001:2181 --describe --topic yiyang
Topic: yiyang PartitionCount: 6 ReplicationFactor: 1 Configs:
Topic: yiyang Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: yiyang Partition: 1 Leader: 1 Replicas: 1 Isr: 1
Topic: yiyang Partition: 2 Leader: 0 Replicas: 0 Isr: 0
Topic: yiyang Partition: 3 Leader: 1 Replicas: 1 Isr: 1
Topic: yiyang Partition: 4 Leader: 0 Replicas: 0 Isr: 0
Topic: yiyang Partition: 5 Leader: 1 Replicas: 1 Isr: 1
[root@hadoop001 kafka_2.12]#
来源：https://www./content-4-865351.html
|
