大数据之 kafka 的安装及基本操作

在流式计算中，Kafka一般用来缓存数据，Kafka是一个分布式消息队列；Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer，消息接受者称为Consumer，此外kafka集群有多个kafka实例组成，每个实例(server)称为broker；无论是kafka集群，还是consumer都依赖于zookeeper集群保存一些meta信息，来保证系统可用性。
kafka的下载地址：http://kafka.apache.org/downloads.html

1.集群的规划

hadoop101                   hadoop102               hadoop103
zk                          zk                      zk
kafka                       kafka                   kafka

安装 kafka 之前必须先要安装 zookeeper，在这里不再演示 zk 怎么安装。

2.集群部署

1）解压安装包

[hw@hadoop101 software]$ tar -zxvf kafka_2.11-0.11.0.0.tgz -C /opt/module/

2）修改解压后的文件名称 (用默认也行，看个人习惯而定)

[hw@hadoop101 module]$ mv kafka_2.11-0.11.0.0/ kafka

3）在/opt/module/kafka目录下创建logs文件夹

[hw@hadoop101 kafka]$ mkdir logs

4）修改配置文件

[hw@hadoop101 kafka]$ cd config/
[hw@hadoop101 config]$ vi server.properties

输入以下内容：

#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/module/kafka/logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=hadoop101:2181,hadoop102:2181,hadoop103:2181

5）配置环境变量

[hw@hadoop101 module]$ sudo vi /etc/profile

#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin

[hw@hadoop101 module]$ source /etc/profile

6）分发安装包

[hw@hadoop101 module]$ scp -r kafka/ hw@hadoop102:/opt/module/ 
[hw@hadoop101 module]$ scp -r kafka/ hw@hadoop103:/opt/module/ 

7）分别在hadoop102和hadoop103上修改配置文件/opt/module/kafka/config/server.properties中的broker.id=1、broker.id=2
注：broker.id不得重复

8）启动集群
依次在hadoop101、hadoop102、hadoop103节点上启动kafka

[hw@hadoop101 kafka]$ bin/kafka-server-start.sh config/server.properties &
[hw@hadoop102 kafka]$ bin/kafka-server-start.sh config/server.properties &
[hw@hadoop103 kafka]$ bin/kafka-server-start.sh config/server.properties &

这里后面用 & 符号就是为了去掉控制台中打印的信息，如果是第一次启动最好不用 & ，因此看不到报错信息。

9）关闭集群

[hw@hadoop101 kafka]$ bin/kafka-server-stop.sh stop
[hw@hadoop102 kafka]$ bin/kafka-server-stop.sh stop
[hw@hadoop103 kafka]$ bin/kafka-server-stop.sh stop

3.Kafka命令行操作

1）查看当前服务器中的所有topic

[hw@hadoop101 kafka]$ bin/kafka-topics.sh --zookeeper hadoop101:2181 --list

2）创建topic

[hw@hadoop101 kafka]$ bin/kafka-topics.sh --zookeeper hadoop101:2181 \
--create --replication-factor 3 --partitions 1 --topic first

选项说明：
--topic 定义topic名
--replication-factor 定义副本数
--partitions 定义分区数

3）删除topic

[hw@hadoop101 kafka]$ bin/kafka-topics.sh --zookeeper hadoop101:2181 \
--delete --topic first

需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。

4）发送消息

[hw@hadoop101 kafka]$ bin/kafka-console-producer.sh \
--broker-list hadoop101:9092 --topic first
>hello world
>lu chang yin

注：hadoop101:9092 这个 IP 和 端口 必须要和 config/server.properties 文件中的 listeners 所设置的值一致： image.png

消费者可以使用默认值。

5）消费消息

[hw@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper hadoop101:2181 --from-beginning --topic first

--from-beginning：会把first主题中以往所有的数据都读取出来。根据业务场景选择是否增加该配置。

6）查看某个Topic的详情

[hw@hadoop101 kafka]$ bin/kafka-topics.sh --zookeeper hadoop101:2181 \
--describe --topic first

最终看下效果图： image.png

好了，今天总结就到这里，希望能帮助到你，至于 API 的操作待后续更新，欢迎留言。