kafka常用命令
一、脚本简介
1、kafka-acls.sh #配置，查看kafka集群鉴权信息
2、kafka-configs.sh #查看，修改kafka配置
3、kafka-console-consumer.sh #消费命令
4、kafka-console-producer.sh #生产命令
5、kafka-consumer-groups.sh #查看消费者组，重置消费位点等
6、kafka-consumer-perf-test.sh #kafka自带消费性能测试命令
7、kafka-mirror-maker.sh #kafka集群间同步命令
8、kafka-preferred-replica-election.sh #重新选举topic分区leader
9、kafka-producer-perf-test.sh #kafka自带生产性能测试命令
10、kafka-reassign-partitions.sh #kafka数据重平衡命令
11、kafka-run-class.sh #kafka执行脚本
12、kafka-server-start.sh #进程启动
13、kafka-server-stop.sh #进程停止
14、kafka-topics.sh #查询topic状态，新建，删除，扩容
二、常用命令大全
###1、集群相关命令###
bin/zookeeper-server-start.sh config/zookeeper.properties & # 启动zookeeper
bin/zookeeper-server-stop.sh # 停止zookeeper
bin/zkServer.sh start
bin/zkServer.sh stop
bin/kafka-server-start.sh $path/server.properties # 前台启动broker
bin/kafka-server-start.sh -daemon $path/server.properties # 后台启动broker
bin/kafka-server-stop.sh # 关闭broker
###2、topic###
bin/kafka-topics.sh --create --zookeeper $zkhost --replication-factor $replCount --partitions $partitionCount --topic $topicName #指定副本数、pritition数创建topic
bin/kafka-topics.sh --zookeeper $zkhost --alter --topic $topicName --partitions $partitionCount #扩容分区、只能扩，不能缩（涉及数据迁移、合并所以不支持））
bin/kafka-topics.sh --zookeeper $zkhost --alter --config.compression.type=gzip --topic $topicName #动态配置topic参数
--config.compression.type=gzip #修改或添加参数配置
--add-config.compression.type=gzip #添加参数配置
--delete-config.compression.type=gzip #删除参数配置
bin/kafka-topics.sh --zookeeper $zkhost --delete --topic $topicName #删除topic、需关注配置文件delete.topic.enble=true,否则无法真正删除
bin/kafka-topics.sh --zookeeper $zkhost --list #查看topic列表
bin/kafka-topics.sh --zookeeper $zkhost --describe #查看所有topic详细信息
--topic $topicname #指定topic查看详细信息
####################topic详细信息####################
Topic:test PartitionCount:3 ReplicationFactor:3 Configs:
Topic: test Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 0,2,1
Topic: test Partition: 1 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
Topic: test Partition: 2 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
#Topic：topic名称
#prititionCount：partition数量
#replicationFactor：副本数量
#configs：其他配置
#Partition：prititonId
#Leader：leader节点
#Replicas：挂载节点
#ISR：存活节点
######################################################
###3、消费者组###
bin/kafka-consumer-groups.sh --bootstrap-server $nodes --list #消费者列表查询(待验证)
bin/kafka-consumer-groups.sh --bootstrap-server $nodes --describe --group $groupName #显示某个消费组的消费详情(待验证)
bin/kafka-consumer-groups.sh --bootstrap-server $nodes --group $groupname --reset-offsets --all-topics --to-earliest --execute # 重设消费者组位移(待验证)
--to-latest --execute # 最新处
--to-offset 2000 --execute # 某个位置
--to-datetime 2019-09-15T00:00:00.000 # 调整到某个时间之后的最早位移
bin/kafka-consumer-groups.sh --zookeeper $zkhost --delete --group $groupname # 删除消费者组
#####消费组的消费详情#######
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
topic名字 分区id 当前已消费的条数 总条数 未消费的条数 消费id 主机ip 客户端id
############################
###4、生产、消费者###
bin/kafka-console-producer.sh --broker-list $nodes --topic $topicName # 使用生产者
bin/kafka-console-consumer.sh --bootstrap-server $nodes --topic $topicName --from-beginning # 使用消费者
--from-beginning #为可选参数，表示要从头消费消息
--from-earliest #从最早的消息开始消费（待验证）
--from-latest #从最新的消息开始消费
--指定offset #从指定的位置开始消费
bin/kafka-console-consumer.sh --bootstrap-server $nodes --topic $topicName --from-beginning --consumer-property group.id=$groupname # 指定groupid
bin/kafka-console-consumer.sh --bootstrap-server $nodes --topic $topicName --from-beginning --partition $partitionId # 指定分区
bin/kafka-console-consumer.sh --bootstrap-server $nodes --topic $topicName --new-consumer --from-beginning --consumer.config config/consumer.properties # 新消费者
bin/kafka-simple-consumer-shell.sh --brist localhost:9092 --topic test --partition 0 --offset 1234 --max-messages 10 # 高级点的用法(不常使用，未验证)
###5、切换leader###
./bin/kafka-preferred-replica-election.sh --zookeeper 10.161.68.123:2181,10.161.68.124:2181,10.161.68.125:2181 #切换leader，此处指定一个node
###6、压测命令###
bin/kafka-producer-perf-test.sh --topic $topicName --num-records 100 --record-size 1 --throughput 100 --producer-props bootstrap.servers=$nodes #（未验证）
###7、迁移分区###
#可先修改保存时间避免大量数据同步#
bin/kafka-configs.sh --zookeeper=$zkhost --alter --entity-type topics --add-config retention.ms=86400000 --entity-name $topicName
#创建json，json格式如下#
{
"version":1, "partitions":[
{
"topic":"__consumer_offsets","partition":0,"replicas":[0,1]},
{
"topic":"__consumer_offsets","partition":1,"replicas":[0,1]},
{
"topic":"__consumer_offsets","partition":2,"replicas":[0,1]},
{
"topic":"__consumer_offsets","partition":3,"replicas":[0,1]},
{
"topic":"__consumer_offsets","partition":4,"replicas":[0,1]},
{
"topic":"__consumer_offsets","partition":5,"replicas":[0,1]},
{
"topic":"__consumer_offsets","partition":6,"replicas":[0,1]}
]}
#执行#
bin/kafka-reassign-partitions.sh --zookeeper $zkhost --reassignment-json-file $jsonFile --execute
#验证#
bin/kafka-reassign-partitions.sh --zookeeper $zkhost --reassignment-json-file $jsonFile --verify
###8、动态参数配置###
bin/kafka-configs.sh --zookeeper=$zkhost --alter --add-config '$具体参数' --entity-type $entitytype --entity-name $entityname #只可修改支持动态配置的参数，其他的报错
#entityType常用的包括以下几个方面，根据entityType的不同前面选择zk 或者 broker
topics 、 users、 brokers、 concumers、 groups 等
kafka-console-consumer.sh 详解
概述
kafka-console-consumer.sh 脚本是一个简易的消费者控制台。该 shell 脚本的功能通过调用 kafka.tools 包下的 ConsoleConsumer 类，并将提供的命令行参数全部传给该类实现。
注意：Kafka 从 2.2 版本开始将 kafka-topic.sh 脚本中的 −−zookeeper 参数标注为 “过时”，推荐使用
−−bootstrap-server参数。若读者依旧使用的是 2.1 及以下版本，请将下述的 --bootstrap-server 参数及其值手动替换为 --zookeeper zk1:2181,zk2:2181,zk:2181。一定要注意两者参数值所指向的集群地址是不同的。
消息消费
表示从 latest 位移位置开始消费该主题的所有分区消息，即仅消费正在写入的消息。
bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic topicName
从开始位置消费
bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --from-beginning --topic topicName
表示从指定主题中有效的起始位移位置开始消费所有分区的消息。
显示key消费
bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --property print.key=true --topic topicName
消费出的消息结果将打印出消息体的 key 和 value。
参数说明
若还需要为你的消息添加其他属性，请参考下述列表。
| 参数 | 值类型 | 说明 | 有效值 |
|---|---|---|---|
| –topic | string | 被消费的topic | |
| –whitelist | string | 正则表达式，指定要包含以供使用的主题的白名单 | |
| –partition | integer | 指定分区除非指定’–offset’，否则从分区结束(latest)开始消费 | |
| –offset | string | 执行消费的起始offset位置默认值:latest | latest earliest |
| –consumer-property | string | 将用户定义的属性以key=value的形式传递给使用者 | |
| –consumer.config | string | 消费者配置属性文件请注意，[consumer-property]优先于此配置 | |
| –formatter | string | 用于格式化kafka消息以供显示的类的名称默认值:kafka.tools.DefaultMessageFormatter | |
| –property | string | 初始化消息格式化程序的属性 | |
| –from-beginning | 从存在的最早消息开始，而不是从最新消息开始 | ||
| –max-messages | integer | 消费的最大数据量，若不指定，则持续消费下去 | |
| –timeout-ms | integer | 在指定时间间隔内没有消息可用时退出 | |
| –skip-message-on-error | 如果处理消息时出错，请跳过它而不是暂停 | ||
| –bootstrap-server | string | 必需(除非使用旧版本的消费者)，要连接的服务器 | |
| –key-deserializer | string | ||
| –value-deserializer | string | ||
| –enable-systest-events | string | 除记录消费的消息外，还记录消费者的生命周期(用于系统测试) | |
| –isolation-level | string | 设置为read_committed以过滤掉未提交的事务性消息,设置为read_uncommitted以读取所有消息,默认值:read_uncommitted | |
| –blacklist | string | 要从消费中排除的主题黑名单 | |
| –csv-reporter-enabled | 如果设置，将启用csv metrics报告器 | ||
| –delete-consumer-offsets | string | 如果指定，则启动时删除zookeeper中的消费者信息 | |
| –metrics-dir | 输出csv度量值,需与[csv-reporter-enable]配合使用 | ||
| –zookeeper | string | 必需(仅当使用旧的使用者时)连接zookeeper的字符串。可以给出多个URL以允许故障转移 |
=
