kafka学习

Posted 2020-11-20 seeasunnyday

　　　　　　　　　　　　　　　　管理kafka

 

 

主题操作

1.在集群里创建一个主题需要用到3个参数。这些参数是必须提供的，尽管有些已经有broker级别的默认值。

 

主题名字，想要创建的主题的名字，主题名字可以包含字母，数字，下划线以及英文状态下破折号和句号。

复制系数，主题的副本数量

分区，主题的分区数量

 

指定主题配置

可以在创建时显式地指定复制系数或者对配置进行覆盖，通过向kafka-topic.sh 传递 --config参数来实现的。

 

主题创建的格式

kafka-topics.sh --zookeeper <zookeeper connect> --create --topic <string> --relipcation-fact <integer> -- partition <integer>

如果为集群指定了基于机架信息的副本分配策略，那么分区的副本会分布在不同的架构上。如果不需要基于架构信息的分配策略，可以指定参数 --disable-rack-aware

 

 

增加分区

kafak-topics.sh --zookeeper <zookeeper connect> --alter --topic <String> --partitions <integer>

只能增加分区，不能删除分区，如果删除分区，会导致数据不一致。

 

删除主题

 

如果一个主题不再使用，只要它还存在集群里面，就会占用一定数量的磁盘空间和文件句柄，把它删除就可以释放被占用的资源。为了能删除主题，broker的delete.topic.enable 参数必须设置为true,如果该参数被设置为false,删除主题的请求会被忽略。

 

kafka-topics.sh --zookeeper <zookeeper connect> --delete --topic <String>

 

列出集群里的所有主题

 

可以使用主题工具列出集群里的所有主题。每个主题占用一行输出，主题之间没有特定的顺序。

kafka-topic.sh --zookeeper <zookeeper connect> --list

 

列出主题详细信息

 

kafka-topic.sh --zookeeper <zookeeper connect> --describe

 

使用topics-with-overrides 参数可以找出所有包含覆盖配置的主题，它只会列出包含了与集群不一样的主题

 

有两个参数可用于找出有问题的分区。

使用--under-replicated-partitions 参数可以列出所有包含不同步副本的分区。

使用 --unavailable-partitions 参数可以列出所有没有首领的分区，这些分区已经处于离线状态，对于生产者和消费者来说是不可用的。

 

消费者群组

在kafka里，有两个地方保存着消费者群组的信息。对于旧版本的消费来说，它们的信息保存在Zookeeper上。对于新版本的消费者来说，它们的信息保存在broker上，kafka-consumer-group.sh 工具可以用于列出上述两种消费者群组。它也可以用于删除消费者群组和偏移量信息，不过这个功能仅限于旧版本的消费者群组(信息保存在Zookeeper上)，在对于旧版本的消费者仅限操作时，需要通过--zookeeper参数来指定Zookeeper的地址；对于新的 则需要指定 --bootstrap-server参数来指定broker的主机名和端口。

 

列出旧版本的消费者群组

 

kafka-consumer-groups.sh --zookeeper <zookeeper connect> --list

 

列出新版本的消费者群组

kafka-consumer-grous.sh --new-consumer --bootstrap-server <kafka connect> --list

 

列出的任意群组来说 使用 --describe 代替 --list.

 

 

消费者群组的详细信息

 

Group 消费者群组的名字

Topic 正在被读取的主题名字

Partition 正在被读取的分区ID

Current-offset 消费者群组最近提交的偏移量，也就是说在分区里去读取的当前位置。

log-end-offset 当前高水位偏移量，也就是最近一被读取消息的偏移量，同时也是最近一个提交到集群的偏移量

lag 消费者的current-offset 和broker的log-end-offset之间的差距

owner 消费者群组里正在读取该分区的消费者。这个一个消费者的id,不一定包含消费者的主机名。

 

消费群组

只有旧版本的消费者客户端才支持删除群组的操作。删除群组操作将从zookeeper上移除整个群组，包括所有已保存的偏移量。

 

kafka-consumer-groups.sh --zookeeper <zookeeper connect> --delete --group <String>

 

该命令也可以用于不删除整个群组的情况下删除单个主题的偏移量，再次强调，在进行删除操作之前，需要先关闭消费者，或者不要让它们读取即将被删除的主题。

kafka-consumer-groups.sh --zookeeper <zookeeper connect> --delete --group <String> --topic <String>

 

偏移量管理

 

1.导出偏移量

kafka-run-class.sh kafka.tools.ExportZkOffsets --zkconnect <zookeeper connect> --group <String> --output -file <String>

2.导入偏移量

先关闭消费者，如果消费者活跃状态，那么可能会将导入的偏移量被覆盖

kafka-run-class.sh kafka.tools.ImportZkOffsets --zkconnet <zookeeper connect> --input-file <String>

 

覆盖主题的默认配置

kafka-configs.sh --zookepper <zookepper connect> --alter --entity-type topic --entity-name <topic name> --and-config <key>=<value>,<key>=<value>

 

覆盖客户端的默认配置

kafka-configs.sh --zookeeper <zookeeper connect> --alter --entity clients --entity-name <client id> and -config <key>=<value>,<key>=<value>

 

列出被覆盖的配置

kafka-config.sh --zookeeper <zookeeper connect> --describe --entity-type topics --entity-name my-topic

 

移除被覆盖的配置

删除主题的某一个配置

kafka-configs.sh --zookeeper <zookeeper connect> -alter --entity-type topics --entity-name <topic name> --delete-config <参数名>

 

分区管理

 

自动首领再均衡

kafka-preferred-replica-election.sh --zookeeper<zookeeper connect>

 

修改分区副本

使用kafka-reassign-partitions.sh

这个工具分为步来修改分区

1.根据broker清单和主题清单生产一组迁移步骤

2.执行这些迁移步骤。

3.第三步是可选的，可以使用生成的迁移步骤验证分区重分配的进度和完成情况。

获取json

kafak-reassgin-partitions.sh --zookeeper<zookeeper connect> --generate --topic-to-move-json-file topics.json --broker -list 0,1

修改执行json

kafka-reassgin-partitions.sh --zookeeper<zookeeper connect> --execute --reassignment-json-file topic.json

 

显示重新分区的情况

kafka-reassgin-partitions.sh --zookeeper<zookeeper connect> --verify --reassignment-json-file topic.json

 

修改复制系数 从上面步骤中获取json，修改复制系数，执行

 

转储日志片段

 

kafka-run-class.sh kafka.tools.DumplogSegments --files <log name>

验证索引文件的正确性使用 --index-sanity-check

 

验证副本

 

kafka-replica-verfycation.sh --broker-list <kafka connect>，<kafka connect> --topic-while-list <主题名>

 

消费者和生产者借助kafka-console-consumer.sh 和kafka-console-producer.sh

 

指定待读取的主题3个可用的参数 --topic --whilelist --blacklist 后面两个跟着一个正则表达式。

 

使用旧版消费者读取单个主题

kafka-console-consumer.sh --zookeeper <zookepper connect> --topic <tipic name>

控制台消费者其他常用配置如下：

 

--formatter className 指定消息格式化器的类名，用于解码消息 默认值是kafka.tools.DefaultFormatter

 

--form-beginning 指定从最旧的偏移量开始读取数据，否则从最新的偏移量开始读取。

--max-message Num 指定在推出之前最多读取NUM个消息。

--partition NUM 指定只读取ID 为NUM的分区(需要新版本的消费者)。

 

读取偏移量主题 使用这个格式化器读取kafka.coordinator.GroupMetadataManager$OFFsetsMessageFormatter

kafka-console-consumer.sh --zookeeper <zookeeper connect> --topic __consumer_offsets --formatter kafka.coordinator.GroupMetadataManager$OFFsetsMessageFormatter

 

控制台生产者

向主题生成两个消息

kafka-console-prodecer.sh --broker-list <kafka connect> --topic <topic name>

 

客户端ACL

命令行工具kafka-acls.sh可以用于处理客户端访问控制相关的问题。

 

取消分配重分配

1.发起重分配请求

2.集群控制器将分区添加到broker上

3.新的broker开始复制分区，直到副本达到同步状态

4.集群控制器从分区副本清单里移除旧的broker.

 

手动删除主题

1.关闭集群里所有的broker

2.删除zookeeper路径/broker/topics/TOPISCNAME，注意要先删除节点下的子节点。

3.删除每个broker的分区目录，这些目录的名字可能是TOPICNAME-NUM 其中NU是指定分区的ID.

4.重启所有的broker.

1.关闭集群里所有的broker

2.删除zookeeper路径/broker/topics/TOPISCNAME，注意要先删除节点下的子节点。

3.删除每个broker的分区目录，这些目录的名字可能是TOPICNAME-NUM 其中NU是指定分区的ID.

4.重启所有的broker.