小惠珠哦

这个屌丝很懒，什么也没留下！

热门标签

进击消息中间件系列（三）：Kafka 中 shell 命令使用

作者：小惠珠哦 | 2024-08-06 08:49:38

踩

kafka.tools.getoffsetshell

点击下方名片，设为星标！

回复“1024”获取2TB学习资源！

前面介绍了 Kafka 的基础概念与架构、单机与集群部署相关的知识点，今天我将详细的为大家介绍 Kafka 中shell 命令使用相关知识，希望大家能够从中收获多多！如有帮助，请点在看、转发支持一波！！！

创建 kafka topic

kafka-topics.sh --zookeeper node01:2181 --create --topic t_Name --partitions 30 --replication-factor 2

注：partitions指定topic分区数，replication-factor指定topic每个分区的副本数。

partitions分区数:

partitions ：分区数，控制topic将分片成多少个log。可以显示指定，如果不指定则会使用broker(server.properties)中的num.partitions配置的数量。

虽然增加分区数可以提供kafka集群的吞吐量、但是过多的分区数或者或是单台服务器上的分区数过多，会增加不可用及延迟的风险。因为多的分区数，意味着需要打开更多的文件句柄、增加点到点的延时、增加客户端的内存消耗。

分区数也限制了consumer的并行度，即限制了并行consumer消息的线程数不能大于分区数。

分区数也限制了producer发送消息是指定的分区。如创建topic时分区设置为1，producer发送消息时通过自定义的分区方法指定分区为2或以上的数都会出错的；这种情况可以通过alter –partitions 来增加分区数。分区数只可以增加不可以减少。

replication-factor副本

replication factor 控制消息保存在几个broker(服务器)上，一般情况下等于broker的个数。

如果没有在创建时显示指定或通过API向一个不存在的topic生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量。更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

查看所有topic列表

kafka-topics.sh --zookeeper node01:2181 --list

查看指定topic的详细信息

kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name

创建一个topic并生产消息


bootstrap_server=hadoop001:9092,hadoop002:9092,hadoop003:9092
zookeeper_connect=hadoop001:2181,hadoop002:2181,hadoop003:2181/kafka
kafka-topics.sh --create --zookeeper $zookeeper_connect --replication-factor 2 --partitions 3 --topic test001
kafka-console-producer.sh --broker-list $bootstrap_server --topic test001 --producer.config $KAFKA_HOME/config/producer.properties


bootstrap_server   #定义的kafka的broker集群节点，端口号在配置文件里可以查看。  
zookeeper_connect  #定义的zookeeper集群节点，kafka是配置文件中指定的对topic进行操作的虚拟用户。

对指定的topic消费


bootstrap_server=hadoop001:9092,hadoop002:9092,hadoop003:9092
zookeeper_connect=hadoop001:2181,hadoop002:2181,hadoop003:2181/kafka
kafka-console-consumer.sh --bootstrap-server $bootstrap_server --from-beginning --topic test001 --consumer.config $KAFKA_HOME/config/consumer.properties

查看topic某分区偏移量最大(小)值

kafka-run-class.sh kafka.tools.GetOffsetShell --topic t_Name --time -1 --broker-list node86:9092 --partitions 0

注：time为-1时表示最大值，time为-2时表示最小值。更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

增加topic分区数

kafka-topics.sh --zookeeper node01:2181 --alter --topic t_cdr --partitions 10

查看topic消费进度

这个会显示出consumer group的offset情况，必须参数为–group，不指定–topic，默认为所有topic。


Displays the: Consumer Group, Topic, Partitions, Offset, logSize, Lag, Owner for the specified set of Topics and Consumer Group
bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker
required argument: [group]
Option Description
------ -----------
--broker-info Print broker info
--group Consumer group.
--help Print this message.
--topic Comma-separated list of consumer
  topics (all topics if absent).
--zkconnect ZooKeeper connect string. (default: localhost:2181)
Example,
kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group pv
Group           Topic              Pid Offset   logSize    Lag    Owner
pv              page_visits        0   21       21         0      none
pv              page_visits        1   19       19         0      none
pv              page_visits        2   20       20         0      none

删除指定的topic

首先在kafka配置文件server.properties添加配置项 delete.topic.enable=true
之后执行命令：kafka-topics.sh --delete --zookeeper 【zookeeper server】 --topic 【topic name】

kafka数据迁移

方法一：通过增加partition数量

kafka-topics.sh --zookeeper localhost:2181 --alter --topic demo1 --partitions 3

方法二：通过重新分配partition

1.编写分配脚本并执行分配计划


vi ~/kafka_add_replicas.json
 
{"topics":
 [{"topic":"prod_log_simul"}],
 "version": 1
 }

之后运行分配计划

kafka-reassign-partitions.sh --zookeeper hdc-data1,hdc-data2,hdc-data3:2181 --topics-to-move-json-file ~/kafka_add_replicas.json --broker-list "0,1,2" --generate


–topics-to-move-json-file：#自己编写的分配脚本  
–broker-list：#中间写的是kafka broker集群的编号，例如：“125,127,10”

2.根据上一步生成的分配计划复制并配置json文件topic-reassignment.json，进行topic的重新分配

vi ~/topic-reassignment.json

之后把Proposed partition reassignment configuration #建议的分区配置 {“version”:1,“partitions”:[{“topic”:“event_request”,“partition”:0,“replicas”:[6,5]},{“topic”:“event_request”,“partition”:1,“replicas”:[7,6]}]} 的内容复制到文件里，然后根据自己的需要修改。

3.执行分配

kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file topic-reassignment.json --execute

执行前的分区分布


[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name
Topic:event_request        PartitionCount:2   ReplicationFactor:2        Configs:
        Topic: event_request      Partition: 0       Leader: 3  Replicas: 3,4 Isr: 3,4
        Topic: event_request      Partition: 1       Leader: 4  Replicas: 4,5 Isr: 4,5

执行后的分区分布


[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name
Topic:event_request        PartitionCount:2   ReplicationFactor:4        Configs:
        Topic: event_request      Partition: 0       Leader: 3  Replicas: 6,5,3,4     Isr: 3,4
        Topic: event_request      Partition: 1       Leader: 4  Replicas: 7,6,4,5     Isr: 4,5

4.检查分配的状态

查看分配的状态：正在进行


[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file topic-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [event_request,0] is still in progress
Reassignment of partition [event_request,1] is still in progress

查看“is still in progress” 状态时的分区，副本分布状态：发现Replicas有4个哦，说明在重新分配的过程中新旧的副本都在进行工作。更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。


[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name       
        PartitionCount:2   ReplicationFactor:4        Configs:
        Topic: event_request      Partition: 0       Leader: 3  Replicas: 6,5,3,4     Isr: 3,4
        Topic: event_request      Partition: 1       Leader: 4  Replicas: 7,6,4,5     Isr: 4,5

查看分配的状态：分配完成。


[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file topic-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [event_request,0] completed successfully
Reassignment of partition [event_request,1] completed successfully

查看“completed successfully”状态的分区，副本状态：已经按照生成的分配计划正确的完成了分区的重新分配。


[hadoop@sdf-nimbus-perf topic_reassgin]$ kafka-topics.sh --zookeeper node01:2181 --describe --topic t_Name Topic:event_request        PartitionCount:2   ReplicationFactor:2        Configs:
        Topic: event_request      Partition: 0       Leader: 6  Replicas: 6,5 Isr: 6,5
        Topic: event_request      Partition: 1       Leader: 7  Replicas: 7,6 Isr: 6,7

kafka-configs.sh的使用–修改参数配置项

脚本语法解析

kafka-configs.sh参数解析

增加配置项

某个topic配置对象

bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name topicName  --add-config 'k1=v1, k2=v2, k3=v3'

所有clientId的配置对象

bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type clients --entity-default --add-config 'k1=v1, k2=v2, k3=v3'

例子


bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name topicName  --add-config 'max.message.bytes=50000000, flush.messages=50000, flush.ms=5000'
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name topicName  --add-config 'max.message.bytes=50000000' --add-config 'flush.messages=50000'

删除配置项


bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name topicName --delete-config ‘k1,k2,k3’
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type clients --entity-name clientId --delete-config ‘k1,k2,k3’
bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type brokers --entity-name $brokerId --delete-config ‘k1,k2,k3’
bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type brokers --entity-default --delete-config ‘k1,k2,k3’
例子
bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --alter --entity-type topics --entity-name test-cqy --delete-config 'segment.bytes'

修改配置项

修改配置项与增加语法格式相同，相同参数后端直接覆盖，列出entity配置描述。


bin/kafka-configs.sh --zookeeper localhost:2181/kafkacluster --entity-type topics --entity-name topicName --describe
bin/kafka-configs.sh--bootstrap-server localhost:9092 --entity-type brokers --entity-name $brokerId --describe
bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-default --describe
bin/kafka-configs.sh  --zookeeper localhost:2181/kafkacluster --entity-type users --entity-name user1 --entity-type clients --entity-name clientA --describe

更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

kafka调整分区副本数

创建配置文件 increase-replication-factor.json，这个配置文件说明，


{
    "partitions": [{
            "topic": "ba_spam_content",
            "partition": 0,
            "replicas": [1, 2]
        },
        {
            "topic": "ba_spam_content",
            "partition": 1,
            "replicas": [0, 2]
        },
        {
            "topic": "ba_spam_content",
            "partition": 2,
            "replicas": [1, 0]
        }
    ],
    "version": 1
}

可以看到，单机的情况，replicas 只在0这个broke上。
开始设置副本数

./bin/kafka-reassign-partitions.sh --zookeeper 127.0.0.1:2181 --reassignment-json-file  increase-replication-factor.json --execute

查看副本情况

./bin/kafka-reassign-partitions.sh --zookeeper 127.0.0.1:2181 --reassignment-json-file  increase-replication-factor.json --verify

查看分区信息

kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic ba_spam_content

验证设置的结果情况，可以看到，我们的结果非常好，都成功了
后面可以看到配置很OK，看到有备份信息了。

bin/kafka自带脚本解释

kafka消费者组查看

不同版本可能命令稍微有不同，根据提示配置相关参数。使用脚本：./kafka-consumer-groups.sh 用于查询偏移量信息，数据积压等信息。


kafka消费者组分两种情况，分别查看，根据版本由0.9为版本划分
0.9之前偏移量由zookeepr管理，由于zk不适合大批量写操作，
0.9之后由kafka自己管理，偏移量信息都保存在__consumer_offsets的topic中

查看zk管理的消费者组

This will only show information about consumers that use ZooKeeper (not those using the Java consumer API) 这将只显示使用ZooKeeper的消费者的信息(而不是使用Java消费者API的消费者)。


./kafka-consumer-groups.sh --new-consumer --zookeeper hdp01:2181,hdp02:2181,hdp03:2181 --list
./kafka-consumer-groups.sh --zookeeper hdp01:2181,hdp02:2181,hdp03:2181 --list

查看kafka管理的消费者组

仅仅查看由java 的API开发的consumer

./kafka-consumer-groups.sh --new-consumer --bootstrap-server hdp01:9092,hdp02:9092,hdp03:9092 --list

查看特定consumer group 消费情况

同样根据新/旧版本的consumer，分别指定bootstrap-server与zookeeper参数:


#zk管理的消费者组查看
./kafka-consumer-groups.sh --zookeeper hdp01:2181,hdp02:2181,hdp03:2181 --group t-1810-1 --describe
#kafka管理的消费者组查看
./kafka-consumer-groups.sh --new-consumer --bootstrap-server hdp01:9092,hdp02:9092,hdp03:9092 --group test --describe
 
查询结果：
LAG:积压的偏移量
LOG-END-OFFSET:接受到的最大偏移
CURRENT:已消费的最大偏移

更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

使用 kafka-run-class.sh查看topic

查看消费进度

和上面查看结果一样。只是方式稍微不同：通过指定类去查看消费者组的详细消费kafka.admin.ConsumerGroupCommand类。


#使用规则：由于新旧版本不同，指定不同的zk,boot去查看消费者组
 ./kafka-run-class.sh
 0.9版本以后类 ：kafka.admin.ConsumerGroupCommand
 0.9版本以前类：kafka.tools.ConsumerOffsetChecker 0.9之后该类貌似已经不存在
  --zookeeper hdp01:2181,hdp02:2181,hdp03:2181 ：#查看zk管理的消费者组
  --bootstrap-server hdp01:9092,hdp02:9092,hdp03:9092 ：#查看kafka自己管理的消费者组
   --group  t-1810-1 
   --describe
 
#用例：
 ./kafka-run-class.sh kafka.admin.ConsumerGroupCommand --zookeeper hdp01:2181,hdp02:2181,hdp03:2181
  --group  t-1810-1 --describe
./kafka-run-class.sh kafka.admin.ConsumerGroupCommand --bootstrap-server hdp01:9092,hdp02:9092,hdp03:9092 --group  t-1810-1 --describe

查看topic最大消息数


kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 194.168.221.17:21005,194.168.221.18:21005,194.168.221.19:21005,194.168.221.20:21005 --topic GQ_RESOURCE --time -1 
 
-1:#表示分区当前生产的最大消息条。总消息数=各分区和
3:#解码存储文件  
   可以用来解码存储的.log和.index文件  
.sh kafka.tools.DumpLogSegments --files name.log – print-data-log

kafka一键启动/关闭脚本

启动脚本


#!/bin/bash
brokers="hdp01 hdp02 hdp03"
KAFKA_HOME="/home/hadoop/apps/kafka_2.11-1.1.0"
KAFKA_NAME="kafka_2.11-1.1.0"
 
echo "INFO : Begin to start kafka cluster ..."
 
for broker in $brokers
do
  echo "INFO : Starting ${KAFKA_NAME} on ${broker} ..."
  ssh ${broker} -C "source /etc/profile; sh ${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties 1>~/logs/kafka_std.log 2>~/logs/kafka_err.log &"
  if [[ $? -eq 0 ]]; then
      echo "INFO:[${broker}] Start successfully"
  fi
done
echo "INFO:Kafka cluster starts successfully !"

停止脚本


#!/bin/bash
brokers="hdp01 hdp02 hdp03"
KAFKA_HOME="/home/hadoop/apps/kafka_2.11-1.1.0"
KAFKA_NAME="kafka_2.11-1.1.0"
 
echo "INFO : Shut down kafka cluster ..."
 
for broker in $brokers
do
  echo "INFO : Starting ${KAFKA_NAME} on ${broker} ..."
  ssh ${broker} -C "source /etc/profile; sh ${KAFKA_HOME}/bin/kafka-server-stop.sh"
  if [[ $? -eq 0 ]]; then
      echo "INFO:[${broker}] stop successfully"
  fi
done
echo "INFO:Kafka cluster stop successfully !"

更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafk，本系列持续更新中。

参考文章：https://blog.csdn.net/weixin_43930865/article

/details/102762677 https://blog.csdn.net/qq_43536701/

article/details/111150285

读者专属技术群

构建高质量的技术交流社群，欢迎从事后端开发、运维技术进群（备注岗位，已在技术交流群的请勿重复添加）。主要以技术交流、内推、行业探讨为主，请文明发言。广告人士勿入，切勿轻信私聊，防止被骗。

扫码加我好友，拉你进群

推荐阅读点击标题可跳转

36 张图，一次性补全你的网络知识短板！

面试官：如何在Docker容器中抓包？问倒一大片

又一款轻量级监控利器！开源了

去了一家不到20人的IT公司后，真的是大开眼界

目前工资最高的外包公司汇总（2023最新版）

卸载 Postman！事实证明，它更牛逼。。。

果真有这样的公司，够横！直接干上了热搜

PS：因为公众号平台更改了推送规则，如果不想错过内容，记得读完点一下“在看”，加个“星标”，这样每次新文章推送才会第一时间出现在你的订阅列表里。点“在看”支持我们吧!

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小惠珠哦/article/detail/936752