Kafka分布式集群安装部署
详尽的视频讲解，请查看该地址：https://edu.csdn.net/course/detail/9572
1.下载
Kafka版本选择：
Flume对Kafka版本的要求：http://flume.apache.org/FlumeUserGuide.html#kafka-sink
Spark对Kafka版本的要求：http://spark.apache.org/docs/2.3.0/structured-streaming-kafka-integration.html
Kafka版本下载：
0.10版本以上下载：http://mirrors.hust.edu.cn/apache/kafka/
所有版本下载地址：http://kafka.apache.org/downloads
这里选择下载kafka_2.11-0.10.2.2.tgz版本，并上传至主节点/home/hadoop/app目录下。
2.解压
使用如下命令解压Kafka安装包：
tar -zxvf kafka_2.11-0.10.2.2.tgz
删除Kafka安装包：
rm -rf kafka_2.11-0.10.2.2.tgz
3. 修改配置文件
3.1修改zookeeper.properties
$ vi zookeeper.properties #Zookeeper 数据存储路径与Zookeeper配置文件保持一致 dataDir=/home/hadoop/data/zookeeper/zkdata
3.2修改consumer.properties
$ vi consumer.properties #配置Zookeeper集群连接地址 zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181
3.3修改producer.properties
$ vi producer.properties #修改kafka集群配置地址（0.10） bootstrap.servers=master:9092,slave1:9092,slave2:9092 #修改kafka集群配置地址（0.8） metadata.broker.list=master:9092,slave1:9092,slave2:9092
3.4修改server.properties
$ vi server.properties #配置Zookeeper集群地址 zookeeper.connect=master:2181,slave1:2181,slave2:2181/kafka1.0 #存储日志文件目录 log.dirs=/home/hadoop/data/kafka-logs
4. Kafka安装目录同步
通过集群分发脚本将kafka整个安装目录分发到其他节点
$ deploy.sh kafka_2.11-0.10.2.2 /home/hadoop/app/ slave
5. 再次修改server.properties
在各个节点分别修改server.properties
#修改hadoop01节点
broker.id=1
#修改hadoop02节点
broker.id=2
#修改hadoop03节点
broker.id=3
6. 创建日志目录
通过远程脚本在各个节点分别创建日志目录：
runRemoteCmd.sh "mkdir /home/hadoop/data/kafka-logs" all
7.Kafka集群启动与测试
7.1启动Zookeeper集群
$ runRemoteCmd.sh "/home/hadoop/app/zookeeper/bin/zkServer.sh start" all
7.2启动Kafka集群
bin/kafka-server-start.sh config/server.properties &
7.3查看topic列表
查看topic列表(如果配了根目录，Zookeeper路径需要加上root目录，例如localhost:2181/kafka1.0)
bin/kafka-topics.sh --zookeeper localhost:2181/kafka1.0 --list
7.4创建topic
bin/kafka-topics.sh --zookeeper localhost:2181/kafka1.0 --create --topic test --replication-factor 3 --partitions 3
7.5查看topic详情
bin/kafka-topics.sh --zookeeper localhost:2181/kafka1.0 --describe --topic test
7.6topic数据发送与消费
7.6.1新api使用
hadoop02使用自带脚本消费topic数据
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
hadoop01使用自带脚本向topic发送数据
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
hadoop03使用自带脚本消费topic数据（此时消费最新数据）
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
hadoop03使用自带脚本消费topic数据（从头消费数据）
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
查看特定consumer group 详情
bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 –list
查看topic每个partition数据消费情况
bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --group xxx –describe
参数说明：
Group 消费者组
TOPIC：曾经消费或正在消费的topic
PARTITION：分区编号
CURRENT-OFFSET：consumer group最后一次提交的offset
LOG-END-OFFSET：最后提交的生产消息offset
LAG：消费offset与生产offset之间的差值
CONSUMER-ID：当前消费topic-partition的group成员id
HOST：消费节点的ip地址
CLIENT-ID：客户端id
7.6.2旧api使用
hadoop02使用自带脚本消费topic数据
bin/kafka-console-consumer.sh --zookeeper localhost:2181/kafka1.0 --topic test
hadoop01使用自带脚本向topic发送数据
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
hadoop03使用自带脚本消费topic数据（此时消费最新数据）
bin/kafka-console-consumer.sh --zookeeper localhost:2181/kafka1.0 --topic test
hadoop03使用自带脚本消费topic数据（从头消费数据）
bin/kafka-console-consumer.sh --zookeeper localhost:2181/kafka1.0 --topic test --from-beginning
查看特定consumer group 详情
bin/kafka-consumer-groups.sh --zookeeper localhost:2181/kafka1.0 --list
查看topic每个partition数据消费情况
bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper localhost:2181 --group xxx
参数说明：
Group：消费者组
Topic：消费的topic名称
pid：分区编号
Offset：消费者组在分区上消费的位置
logSize：分区上的累计消息条数
Lag：消费offset与生产offset之间的差值
Owner：线程名称。
8.Kafka集群监控
KafkaOffsetMonitor是一个可以用于监控Kafka的Topic及Consumer消费状况的工具。以程序一个jar包的形式运行，部署较为方便。只有监控功能，使用起来也较为安全。
作用：
1.监控Kafka集群状态，Topic、Consumer Group列表。
2.图形化展示topic和Consumer之间的关系。
3.图形化展示Consumer的offset、Lag等信息。
8.1下载
下载地址：https://github.com/quantifind/KafkaOffsetMonitor（可以使用已经修改版本）
目前kafka Monitor必须使用旧api才可以监控到，新api目前还没有实现。
8.2脚本参数格式
java -cp KafkaOffsetMonitor-assembly-0.2.1.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk hadoop01:2181,hadoop02:2181,hadoop03:2181/kafka1.0 --port 8090 --refresh 10.seconds --retain 2.days
zk：Zookeeper集群地址
port：为开启web界面的端口号
refresh：刷新时间
retain：数据保留时间（单位seconds, minutes, hours, days）
8.3开发kafkamonitor.sh执行脚本
vi kafkamonitor.sh
#!/bin/sh
home=$(cd `dirname $0`; cd ..; pwd)
. ${home}/bin/common.sh
java -cp ${lib_home}KafkaOffsetMonitor-assembly-0.2.0.jar
com.quantifind.kafka.offsetapp.OffsetGetterWeb
--zk hadoop01:2181,hadoop02:2181,hadoop03:2181/kafka1.0
--port 8090
--refresh 10.seconds
--retain 2.days > /dev/null 2>&1 &
8.4脚本授权
给脚本kafkamonitor.sh赋予可执行权限
chmod u+x kafkamonitor.sh
8.5启动监控脚本
bin/kafkamonitor.sh
8.6web可视化
hadoop01:8090
