关于kafka:kafka运维Kafka全网最全最详细运维命令合集精品强烈建议收藏

首发公众号: 石臻臻的杂货铺 ID：jjdlmn
集体 wx: jjdlmn_

以下大部分运维操作, 都能够应用 LogI-Kafka-Manager 在平台上可视化操作;

@[TOC]

1.TopicCommand

1.1.Topic 创立

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic test

相干可选参数

1.2. 删除 Topic

bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test

反对正则表达式匹配 Topic 来进行删除, 只须要将 topic 用双引号包裹起来
例如: 删除以 create_topic_byhand_zk 为结尾的 topic;

bin/kafka-topics.sh –bootstrap-server localhost:9092 –delete –topic “create_topic_byhand_zk.*”
.示意任意匹配除换行符 \n 之外的任何单字符。要匹配 .，请应用 .。
·*·：匹配后面的子表达式零次或屡次。要匹配 * 字符，请应用 *。
.* : 任意字符

删除任意 Topic (慎用)

bin/kafka-topics.sh –bootstrap-server localhost:9092 –delete –topic “.*?”

更多的用法请参考正则表达式

1.3.Topic 分区扩容

zk 形式(不举荐)

>bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic topic1 --partitions 2

kafka 版本 >= 2.2 反对上面形式（举荐）

单个 Topic 扩容

bin/kafka-topics.sh --bootstrap-server broker_host:port --alter --topic test_create_topic1 --partitions 4

批量扩容 (将所有正则表达式匹配到的 Topic 分区扩容到 4 个)

sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server 172.23.248.85:9092 --alter --partitions 4

".*?" 正则表达式的意思是匹配所有; 您可按需匹配

PS: 当某个 Topic 的分区少于指定的分区数时候, 他会抛出异样; 然而不会影响其余 Topic 失常进行;

相干可选参数

PS: 尽管这里配置的是全副的分区正本调配配置, 然而正在失效的是新增的分区;
比方: 以前 3 分区 1 正本是这样的

当初新增一个分区,--replica-assignment 2,1,3,4 ; 看这个意思如同是把 0，1 号分区相互换个 Broker

然而实际上不会这样做,Controller 在解决的时候会把后面 3 个截掉; 只取新增的分区调配形式, 原来的还是不会变

1.4. 查问 Topic 形容

1. 查问单个 Topic

sh bin/kafka-topics.sh --topic test --bootstrap-server xxxx:9092 --describe --exclude-internal

2. 批量查问 Topic(正则表达式匹配, 上面是查问所有 Topic)
sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server xxxx:9092 --describe --exclude-internal

反对正则表达式匹配 Topic, 只须要将 topic 用双引号包裹起来

相干可选参数

5. 查问 Topic 列表

1. 查问所有 Topic 列表

sh bin/kafka-topics.sh --bootstrap-server xxxxxx:9092 --list --exclude-internal

2. 查问匹配 Topic 列表(正则表达式)

查问 test_create_ 结尾的所有 Topic 列表
sh bin/kafka-topics.sh --bootstrap-server xxxxxx:9092 --list --exclude-internal --topic "test_create_.*"

相干可选参数

2.ConfigCommand

Config 相干操作; 动静配置能够笼罩默认的动态配置;

2.1 查问配置

Topic 配置查问

展现对于 Topic 的动动态配置

1. 查问单个 Topic 配置(只列举动静配置)

sh bin/kafka-configs.sh --describe --bootstrap-server xxxxx:9092 --topic test_create_topic
或者
sh bin/kafka-configs.sh --describe --bootstrap-server 172.23.248.85:9092 --entity-type topics --entity-name test_create_topic

2. 查问所有 Topic 配置(包含外部 Topic)(只列举动静配置)
sh bin/kafka-configs.sh --describe --bootstrap-server 172.23.248.85:9092 --entity-type topics

3. 查问 Topic 的具体配置(动静 + 动态)

只须要加上一个参数--all

其余配置 /clients/users/brokers/broker-loggers 的查问

同理；只须要将--entity-type 改成对应的类型就行了 (topics/clients/users/brokers/broker-loggers)

查问 kafka 版本信息

sh bin/kafka-configs.sh --describe --bootstrap-server xxxx:9092 --version

<font color=red> 所有可配置的动静配置 请看最初面的 附件 局部 </font>

2.2 增删改 配置 --alter

–alter

删除配置 : --delete-config k1=v1,k2=v2
增加 / 批改配 置: --add-config k1,k2
抉择类型: --entity-type (topics/clients/users/brokers/broker-

                                     loggers)

类型名称: --entity-name

Topic 增加 / 批改动静配置

--add-config

sh bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type topics --entity-name test_create_topic1 --add-config file.delete.delay.ms=222222,retention.ms=999999

Topic 删除动静配置

--delete-config

sh bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type topics --entity-name test_create_topic1 --delete-config file.delete.delay.ms,retention.ms

其余配置同理, 只须要类型改下--entity-type

类型有: (topics/clients/users/brokers/broker- loggers)

<font color=red> 哪些配置能够批改 请看最初面的附件：ConfigCommand 的一些可选配置 </font>

3. 正本扩缩、分区迁徙、跨门路迁徙 kafka-reassign-partitions

请戳 [【kafka 运维】正本扩缩容、数据迁徙、正本重调配、正本跨门路迁徙]() (如果点不进去, 示意文章暂未发表, 请急躁期待)

4.Topic 的发送 kafka-console-producer.sh

4.1 生产无 key 音讯

## 生产者
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config config/producer.properties

4.2 生产有 key 音讯
加上属性--property parse.key=true

## 生产者
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config config/producer.properties  --property parse.key=true

<font color=red> 默认音讯 key 与音讯 value 间应用“Tab 键”进行分隔，所以音讯 key 以及 value 中切勿应用转义字符(\t)</font>

可选参数

5. Topic 的生产 kafka-console-consumer.sh

1. 新客户端从头生产--from-beginning (留神这里是新客户端, 如果之前曾经生产过了是不会从头生产的)
上面没有指定客户端名称, 所以每次执行都是新客户端都会从头生产

sh bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –from-beginning

2. 正则表达式匹配 topic 进行生产 --whitelist
生产所有的 topic

sh bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –whitelist ‘.*’

生产所有的 topic，并且还从头生产
sh bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –whitelist ‘.*’ –from-beginning

3. 显示 key 进行生产--property print.key=true

sh bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –property print.key=true

4. 指定分区生产--partition 指定起始偏移量生产--offset

sh bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –partition 0 –offset 100

5. 给客户端命名--group

留神给客户端命名之后, 如果之前有过生产，那么 --from-beginning 就不会再从头生产了

sh bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –group test-group

6. 增加客户端属性--consumer-property

这个参数也能够给客户端增加属性, 然而留神 不能多个中央配置同一个属性, 他们是互斥的; 比方在上面的根底上还加上属性--group test-group 那必定不行

sh bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test --consumer-property group.id=test-consumer-group

7. 增加客户端属性--consumer.config

跟--consumer-property 一样的性质, 都是增加客户端的属性, 不过这里是指定一个文件, 把属性写在文件外面, --consumer-property 的优先级大于 --consumer.config

sh bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –consumer.config config/consumer.properties

6.kafka-leader-election Leader 从新选举

6.1 指定 Topic 指定分区用从新PREFERRED：优先正本策略 进行 Leader 重选举

> sh bin/kafka-leader-election.sh --bootstrap-server xxxx:9090 --topic test_create_topic4 --election-type PREFERRED --partition 0

6.2 所有 Topic 所有分区用从新PREFERRED：优先正本策略 进行 Leader 重选举

sh bin/kafka-leader-election.sh --bootstrap-server xxxx:9090 --election-type preferred  --all-topic-partitions

6.3 设置配置文件批量指定 topic 和分区进行 Leader 重选举

先配置 leader-election.json 文件

{
  "partitions": [
    {
      "topic": "test_create_topic4",
      "partition": 1
    },
    {
      "topic": "test_create_topic4",
      "partition": 2
    }
  ]
}

 sh bin/kafka-leader-election.sh --bootstrap-server xxx:9090 --election-type preferred  --path-to-json-file config/leader-election.json
 

相干可选参数

7. 继续批量推送音讯 kafka-verifiable-producer.sh

单次发送 100 条音讯--max-messages 100

一共要推送多少条，默认为 -1，- 1 示意始终推送到过程敞开地位

sh bin/kafka-verifiable-producer.sh –topic test_create_topic4 –bootstrap-server localhost:9092 --max-messages 100

每秒发送最大吞吐量不超过音讯 --throughput 100

推送音讯时的吞吐量，单位 messages/sec。默认为 -1，示意没有限度

sh bin/kafka-verifiable-producer.sh –topic test_create_topic4 –bootstrap-server localhost:9092 --throughput 100

发送的音讯体带前缀--value-prefix

sh bin/kafka-verifiable-producer.sh –topic test_create_topic4 –bootstrap-server localhost:9092 --value-prefix 666

留神 --value-prefix 666 必须是整数, 发送的音讯体的格局是加上一个 点号. 例如：666.

其余参数：
--producer.config CONFIG_FILE 指定 producer 的配置文件
--acks ACKS 每次推送音讯的 ack 值，默认是 -1

8. 继续批量拉取音讯 kafka-verifiable-consumer

继续生产

sh bin/kafka-verifiable-consumer.sh –group-id test_consumer –bootstrap-server localhost:9092 –topic test_create_topic4

单次最大生产 10 条音讯--max-messages 10

sh bin/kafka-verifiable-consumer.sh –group-id test_consumer –bootstrap-server localhost:9092 –topic test_create_topic4 --max-messages 10

相干可选参数

9. 生产者压力测试 kafka-producer-perf-test.sh

1. 发送 1024 条音讯 --num-records 100 并且每条音讯大小为 1KB--record-size 1024 最大吞吐量每秒 10000 条--throughput 100

sh bin/kafka-producer-perf-test.sh –topic test_create_topic4 –num-records 100 –throughput 100000 –producer-props bootstrap.servers=localhost:9092 –record-size 1024

你能够通过 LogIKM 查看分区是否减少了对应的数据大小

从 LogIKM 能够看到发送了 1024 条音讯; 并且总数据量 =1M; 1024 条 *1024byte = 1M;

2. 用指定音讯文件 --payload-file 发送 100 条音讯最大吞吐量每秒 100 条--throughput 100

1. 先配置好消息文件batchmessage.txt
   

2. 而后执行命令
   发送的音讯会从batchmessage.txt 外面随机抉择; 留神这里咱们没有用参数 --payload-delimeter 指定分隔符，默认分隔符是 \n 换行;

   bin/kafka-producer-perf-test.sh –topic test_create_topic4 –num-records 100 –throughput 100 –producer-props bootstrap.servers=localhost:9090 –payload-file config/batchmessage.txt

3. 验证音讯，能够通过 LogIKM 查看发送的音讯

相干可选参数

10. 消费者压力测试 kafka-consumer-perf-test.sh

生产 100 条音讯 --messages 100

sh bin/kafka-consumer-perf-test.sh -topic test_create_topic4 –bootstrap-server localhost:9090 –messages 100

相干可选参数

11. 删除指定分区的音讯 kafka-delete-records.sh

删除指定 topic 的某个分区的音讯删除至 offset 为 1024

先配置 json 文件offset-json-file.json

{"partitions":
[{"topic": "test1", "partition": 0,
  "offset": 1024}],
  "version":1
}

在执行命令

sh bin/kafka-delete-records.sh –bootstrap-server 172.23.250.249:9090 –offset-json-file config/offset-json-file.json

验证 通过 LogIKM 查看发送的音讯

从这里能够看进去, 配置"offset": 1024 的意思是从最开始的中央删除音讯到 1024 的 offset; 是从最后面开始删除的

12. 查看 Broker 磁盘信息

查问指定 topic 磁盘信息 --topic-list topic1,topic2

sh bin/kafka-log-dirs.sh –bootstrap-server xxxx:9090 –describe –topic-list test2

查问指定 Broker 磁盘信息--broker-list 0 broker1,broker2

sh bin/kafka-log-dirs.sh –bootstrap-server xxxxx:9090 –describe –topic-list test2 –broker-list 0

例如我一个 3 分区 3 正本的 Topic 的查出来的信息
logDir Broker 中配置的log.dir

{
    "version": 1,
    "brokers": [{
        "broker": 0,
        "logDirs": [{
            "logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-0",
            "error": null,
            "partitions": [{
                "partition": "test2-1",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }, {
                "partition": "test2-0",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }, {
                "partition": "test2-2",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }]
        }]
    }, {
        "broker": 1,
        "logDirs": [{
            "logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-1",
            "error": null,
            "partitions": [{
                "partition": "test2-1",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }, {
                "partition": "test2-0",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }, {
                "partition": "test2-2",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }]
        }]
    }, {
        "broker": 2,
        "logDirs": [{
            "logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-2",
            "error": null,
            "partitions": [{
                "partition": "test2-1",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }, {
                "partition": "test2-0",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }, {
                "partition": "test2-2",
                "size": 0,
                "offsetLag": 0,
                "isFuture": false
            }]
        }]
    }, {
        "broker": 3,
        "logDirs": [{
            "logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-3",
            "error": null,
            "partitions": []}]
    }]
}

如果你感觉通过命令查问磁盘信息比拟麻烦，你也能够通过 LogIKM 查看

12. 消费者组治理 kafka-consumer-groups.sh

1. 查看消费者列表--list

sh bin/kafka-consumer-groups.sh --bootstrap-server xxxx:9090 --list

先调用 MetadataRequest 拿到所有在线 Broker 列表
再给每个 Broker 发送 ListGroupsRequest 申请获取 消费者组数据

2. 查看消费者组详情--describe

DescribeGroupsRequest

查看生产组详情--group 或 --all-groups

查看指定生产组详情--group
sh bin/kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --group test2_consumer_group

---

查看所有生产组详情 --all-groups
sh bin/kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --all-groups
查看该生产组 生产的所有 Topic、及所在分区、最新生产 offset、Log 最新数据 offset、Lag 还未生产数量、消费者 ID 等等信息

查问消费者成员信息--members

所有生产组成员信息
sh bin/kafka-consumer-groups.sh --describe --all-groups --members --bootstrap-server xxx:9090
指定生产组成员信息
sh bin/kafka-consumer-groups.sh --describe --members --group test2_consumer_group --bootstrap-server xxxx:9090

查问消费者状态信息--state

所有生产组状态信息
sh bin/kafka-consumer-groups.sh --describe --all-groups --state --bootstrap-server xxxx:9090
指定生产组状态信息
sh bin/kafka-consumer-groups.sh --describe --state --group test2_consumer_group --bootstrap-server xxxxx:9090

3. 删除消费者组--delete

DeleteGroupsRequest

删除生产组 –delete

删除指定生产组 --group
sh bin/kafka-consumer-groups.sh --delete --group test2_consumer_group --bootstrap-server xxxx:9090
删除所有生产组--all-groups
sh bin/kafka-consumer-groups.sh --delete --all-groups --bootstrap-server xxxx:9090

PS: 想要删除生产组前提是这个生产组的所有客户端都进行生产 / 不在线才可能胜利删除; 否则会报上面异样

Error: Deletion of some consumer groups failed:
* Group 'test2_consumer_group' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty.

4. 重置生产组的偏移量 --reset-offsets

<font color=red> 可能执行胜利的一个前提是 生产组这会是不可用状态;</font>

上面的示例应用的参数是: --dry-run ; 这个参数示意预执行, 会打印进去将要解决的后果;
等你想真正执行的时候请换成参数--excute ;

上面示例 重置模式都是 --to-earliest 重置到最早的;

请依据须要参考上面 相干重置 Offset 的模式 换成其余模式;

重置指定生产组的偏移量 --group

重置指定生产组的所有 Topic 的偏移量 --all-topic
sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --all-topic
重置指定生产组的指定 Topic 的偏移量--topic
sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --topic test2

重置所有生产组的偏移量 --all-group

重置所有生产组的所有 Topic 的偏移量 --all-topic
sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server xxxx:9090 --dry-run --all-topic
重置所有生产组中指定 Topic 的偏移量--topic
sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server xxxx:9090 --dry-run --topic test2

--reset-offsets 前面须要接 重置的模式

相干重置 Offset 的模式

--from-file着重解说一下

下面其余的一些模式重置的都是匹配到的所有分区; 不可能每个分区重置到不同的 offset；不过 --from-file 能够让咱们更灵便一点;

1. 先配置 cvs 文档
   格局为: Topic: 分区号: 重置指标偏移量

   test2,0,100
   test2,1,200
   test2,2,300

2. 执行命令

   sh bin/kafka-consumer-groups.sh --reset-offsets --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --from-file config/reset-offset.csv

5. 删除偏移量delete-offsets

<font color=red> 可能执行胜利的一个前提是 生产组这会是不可用状态;</font>

偏移量被删除了之后,Consumer Group 下次启动的时候, 会从头生产;

sh bin/kafka-consumer-groups.sh --delete-offsets --group test2_consumer_group2 --bootstrap-server XXXX:9090 --topic test2

相干可选参数

附件

ConfigCommand 的一些可选配置

Topic 相干可选配置

Broker 相干可选配置

Users 相干可选配置

clients 相干可选配置

以上大部分运维操作, 都能够应用 LogI-Kafka-Manager 在平台上可视化操作;