Hdoop学习笔记（HDP）-Part.19 安装Kafka

本文介绍: 选择 kafka进行安装选择hdp03-05作为kafka的broker。

目录
Part.01 关于HDP
Part.02 核心组件原理
 Part.03 资源规划
 Part.04 基础环境配置
 Part.05 Yum源配置
 Part.06 安装OracleJDK
Part.07 安装MySQL
Part.08 部署Ambari集群
 Part.09 安装OpenLDAP
Part.10 创建集群
 Part.11 安装Kerberos
Part.12 安装HDFS
Part.13 安装Ranger
Part.14 安装YARN+MR
Part.15 安装HIVE
Part.16 安装HBase
Part.17 安装Spark2
Part.18 安装Flink
Part.19 安装Kafka
Part.20 安装Flume

十九、安装Kaf k a

1.安装 kafka

选择kafka进行安装
在这里插入图片描述
选择hdp03-05作为kafka的broker

Lo g direct o r i es:/data01/k af k a–log s

2.安装kafka manager

下载链接：
https://gitcode.net/mirrors/yahoo/kafka-manager/-/archive/master/kafka-manager-master.tar.gz
在具有 sbt 编译环境的服务器（需要外网环境）上，对kafka–manager进行编译

unzip kafka-manager-2.0.0.2.zip -d /usr/local/

修改配置文件，/usr/local/kafka–manager-2.0.0.2/conf/application.conf
修改zookeeper 连接地址

kafka-manager.zkhosts="hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181"

启动kafka-manager

cd /usr/local/kafka-manager-2.0.0.2/bin/
nohup bin/kafka-manager &gt;/dev/null 2&gt;&amp;1 &amp;

访问页面 http://h dp01.h dp.com:9000
创建kafka集群
在这里插入图片描述
Cluste r Nam e：h dp315，自定义名称
Cluste r Zo o keep er Ho sts：h dp01.h dp.com:2181,h dp02.h dp02.com:2181,h dp03.h dp.co m:2181

3.启用 kerberos 配置 确认

(1)Zo oKee per 开启 kerberos 认证

Zo oKee per 账号信息

klist -ket /etc/security/keytabs/zk.service.keytab

在这里插入图片描述
jaas 认证信息
/usr/h d p/3.1.5.0-152/zookeeper/conf/zookeeper _jaa s.conf
/usr/hd p/3.1.5.0-152/zookeeper/conf/zookeeper _client _jaa s.co nf
co m.sun.security.auth.module.Krb5Lo ginModule 这个是类名,是ke rberos对JAAS中的LoginModule的实现;required 表示必须进行校验;其他几个是kerberos 相关参数;后面会将该配置文件地址配置到jvm 参数,其中的信息会被初始化到LoginCon text 上下文对象中。另外注意格式以及末尾的;分号。
在这里插入图片描述
开启 sasl/kerberos 认证，/usr/hd p/3.1.5.0-152/zookeeper/co nf/zoo.cfg

(2)Kafka账号信息

Ambari 自动创建的kafka账号信息如下

klist -kte /etc/security/keytabs/kafka.service.keytab

在这里插入图片描述

(3)Kafka服务端 配置

jaas认证信息，/usr/hd p/3.1.5.0-152/kafka/config/kafka_jaas.co nf
在这里插入图片描述
KafkaServer：Kafka服务端
KafkaClient：Kafka客户端
Client：ZooKee per 客户端
Kerberos 认证有两种方式，即利用票证缓存和指定 key t ab。
要使用存储在票证缓存中的Kerberos票证：

sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useTicketCache=true;

要使用 key t ab：

sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="/etc/security/keytabs/alice.keytab" principal="alice@EXAMPLE.COM";

开启 sasl/kerberos认证，/usr/hd p/3.1.5.0-152/kafka/config/server.properties
核心配置如下

advertised.listeners=SASL_PLAINTEXT://hdp03.hdp.com:6667
listeners=SASL_PLAINTEXT://hdp03.hdp.com:6667
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=kafka
sasl.mechanism.inter.broker.protocol=GSSAPI
security.inter.broker.protocol=SASL_PLAINTEXT

(4)Kafka客户端 配置

jaas认证信息，/usr/hdp/3.1.5.0-152/kafka/config/kafka_client _jaas.c onf
在这里插入图片描述
开启 sasl/kerberos认证，新建文件/root/client.properties

security.protocol=SASL_PLAINTEXT
sasl.kerberos.service.name=kafka
sasl.mechanism=GSSAPI

5.其他配置 确认

(1)权限 控制

Kafka本身自带有一个授权的类kafka.security.auth.SimpleAclAuthorizer，可在server.properties 配置，分为Acl和Ranger两种方式。
Acl控制

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

Ranger控制

authorizer.class.name=org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer

当添加权限控制后，会在zk中创建2个节点
节点1：存储ACL信息节点kafka-acl
节点2：存储ACL变更信息节点kafka-acl–changes

(2)listeners/ad verti sed.li steners

内网发布地址用listeners，对外网发布地址时用ad verti sed.listeners

6.常用 指令

创建topic

./kafka-topics.sh --create --zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 --replication-factor 3 --partitions 3 --topic test-topic

partitions指定topic 分区数，控制 topic将分片成多少个log。可以显示指定，如果不指定则会使用 br oker(server.properties)中的num.partitions配置的数量
虽然增加分区数可以提供kafka集群的吞吐量、但是过多的分区数或者或是单台服务器上的分区数过多，会增加不可用及延迟的风险。因为多的分区数，意味着需要打开更多的文件句柄、增加点到点的延时、增加客户端的内存消耗；分区数也限制了consumer的并行度，即限制了并行 consumer 消息的线程数不能大于分区数；分区数也限制了producer 发送消息是指定的分区。如创建topic时分区设置为1，producer 发送消息时通过自定义的分区方法指定分区为2或以上的数都会出错的；这种情况可以通过alter –partitions 来增加分区数。
replication–factor指定topic 每个分区的副本数,控制消息保存在几个broker(服务器)上，一般情况下等于 broker的个数。如果没有在创建时显示指定或通过API向一个不存在的topic 生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量。
查看所有top ic 列表

./kafka-topics.sh  --zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 --list

查看指定topic信息

./kafka-topics.sh --zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 --describe --topic test-topic

控制台向topic生产数据

./kafka-console-producer.sh --broker-list hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 --topic test-topic

控制台消费topic的数据

./kafka-console-consumer.sh --zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 --topic test-topic --from-beginning

增加topic分区数

./kafka-topics.sh --zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 --alter --topic test-topic --partitions 10

删除topic

./kafka-topics.sh --delete --zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 --topic test-topic

只会删除zookeeper中的元数据，消息文件须手动删除
在zookeeper中，以kafka登录后删除对应的文件

kinit -kt /etc/security/keytabs/kafka.service.keytab kafka/hdp03.hdp.com@HDP315.COM
/usr/hdp/3.1.5.0-152/zookeeper/bin/zkCli.sh -server hdp01:2181,hdp02:2181,hdp03:2181
rmr /admin/delete_topics/test-topic
rmr /brokers/topics/test-topic

查看topic消费进度

./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group group1
显示出consumer group的offset情况，必须参数为--group，不指定--topic，默认为所有topic

查看topic某分区偏移量最大（小）值

./kafka-run-class.sh kafka.tools.GetOffsetShell --topic test-topic --time -1 --broker-list hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 --partitions 0

time为-1时表示最大值，time为-2时表示最小值
列出所有topic的用户组列表

./kafka-consumer-groups.sh --bootstrap-server hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 --list

7.实验：发布 订阅

以hdp03作为producer，hdp04作为consumer
【hdp03】上执行启动生产者，并输入随机字符

kinit -kt /etc/security/keytabs/kafka.service.keytab kafka/hdp03.hdp.com@HDP315.COM
./kafka-console-producer.sh --broker-list hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 --topic test-topic --producer.config /root/producer.properties

【hdp04】上执行启动消费者，观察hdp03上输入字符后在hdp04上显示的内容

kinit -kt /etc/security/keytabs/kafka.service.keytab kafka/hdp03.hdp.com@HDP315.COM
./kafka-console-producer.sh --broker-list
./kafka-console-consumer.sh --bootstrap-server hdp03.hdp.com:6667 --topic test-topic --consumer.config /root/consumer.properties

虽然增加分区数可以提供kafka集群的吞吐量、但是过多的分区数或者或是单台服务器上的分区数过多，会增加不可用及延迟的风险。因为多的分区数，意味着需要打开更多的文件句柄、增加点到点的延时、增加客户端的内存消耗；分区数也限制了con sumer的并行度，即限制了并行consumer消息的线程数不能大于分区数；分区数也限制了producer 发送消息是指定的分区。如创建topic时分区设置为1，pro du cer发送消息时通过自定义的分区方法指定分区为2或以上的数都会出错的；这种情况可以通过alter –partitions 来增加分区数。
replication-factor指定topic每个分区的副本数,控制消息保存在几个broker(服务器)上，一般情况下等于broker的个数。如果没有在创建时显示指定或通过API向一个不存在的topic生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量。
查看所有topic列表
./kafka-topics.sh –zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 –list
查看指定topic信息
./kafka-topics.sh –zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 —des cribe –topic test-topic
控制台向topic生产数据
./kafka-console-pro du cer.sh –broker-list hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 –topic test-topic
控制台消费topic的数据
./kafka-console-consumer.sh –zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 –topic test-topic –from-beginning
增加topic分区数
./kafka-topics.sh –zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 —alter –topic test-topic –par titions 10
删除topic
./kafka-topics.sh —delete –zookeeper hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181 –topic test-topic
只会删除zookeeper中的元数据，消息文件须手动删除
在zookeeper中，以kafka登录后删除对应的文件
kinit -kt /etc/security/keytabs/kafka.service.keytab kafka/hdp03.hdp.com@HDP315.COM
/usr/hdp/3.1.5.0-152/zookeeper/bin/zkCli.sh –server hdp01:2181,hdp02:2181,hdp03:2181
rmr /admin/delete_topics/test-topic
rmr /brokers/topics/test-topic
查看topic消费进度
./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker —group group1
显示出consumer group的offset情况，必须参数为–group，不指定–topic，默认为所有topic
查看topic某分区偏移量最大（小）值
./kafka-run-class.sh kafka.tools.GetOffsetShell –topic test-topic —time -1 –broker-list hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 –par titions 0
time为-1时表示最大值，time为-2时表示最小值
 列出所有topic的用户组列表
./kafka-consumer-groups.sh —bootstrap–server hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 –list

7.实验：发布订阅
以hdp03作为produ cer，hdp04作为consumer
【hdp03】上执行启动生产者，并输入随机字符
kinit -kt /etc/security/keytabs/kafka.service.keytab kafka/hdp03.hdp.com@HDP315.COM
./kafka-console-produ cer.sh –broker-list hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 –topic test-topic —producer.config /root/producer.properties
【hdp04】上执行启动消费者，观察hdp03上输入字符后在hdp04上显示的内容
kinit -kt /etc/security/keytabs/kafka.service.keytab kafka/hdp03.hdp.com@HDP315.COM
./kafka-console-producer.sh –broker-list
./kafka-console-consumer.sh —bootstrap–server hdp03.hdp.com:6667 –topic test-topic –consumer.config /root/consumer.properties