本文我们探讨几种关于如何删除kafka主题数据的策略。
场景分析
在进入主题之前,先讨论下需要删除kafka主题数据的应用场景。
场景介绍
kafka消息在过了保留周期之后会被自动清除。但总有一些情况,需要立刻删除消息。
假设这样场景:已经开始给kafka主题生产消息的应用发现了缺陷,接着bug修复程序需要更新,这是kafka主题中已经了一些错误的消息。这样场景通常在开发环境,我们需要的就是快速批量删除这些消息。
模拟环境
为了模拟环境,首先在kafka目录中创建 purge-scenario主题:
$ bin/kafka-topics.sh \
--create --topic purge-scenario --if-not-exists \
--partitions 2 --replication-factor 1 \
--zookeeper localhost:2181
接着使用shuf命令生成随机数,然后通过kafka-console-producer.sh发送kafka主题:
$ /usr/bin/shuf -i 1-100000 -n 50000000 \
| tee -a /tmp/kafka-random-data \
| bin/kafka-console-producer.sh \
--bootstrap-server=0.0.0.0:9092 \
--topic purge-scenario
shuf -i 1-100000 -n 50000000 :表示生成n个1-100000范围内随机数。
tee -a 前面命令结果写入文件 -a 表示追加;这是使用tee保存模拟数据是为了以后使用;
最后验证消费主题消息:
$ bin/kafka-console-consumer.sh \
--bootstrap-server=0.0.0.0:9092 \
--from-beginning --topic purge-scenario \
--max-messages 3
76696
49425
1744
Processed a total of 3 messages
消息过期
在purge-scenario主题中的消息有缺省7天保留时间。为了删除消息,我们可以临时设置主题的 retention.ms 属性为10秒,然后等待其自动过期:
$ bin/kafka-configs.sh --alter \
--add-config retention.ms=10000 \
--bootstrap-server=0.0.0.0:9092 \
--topic purge-scenario \
&& sleep 10
现在验证消费是否过期:
$ bin/kafka-console-consumer.sh \
--bootstrap-server=0.0.0.0:9092 \
--from-beginning --topic purge-scenario \
--max-messages 1 --timeout-ms 1000
[2021-02-28 11:20:15,951] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.TimeoutException
Processed a total of 0 messages
最后,我们要恢复主题的保留周期:
$ bin/kafka-configs.sh --alter \
--add-config retention.ms=604800000 \
--bootstrap-server=0.0.0.0:9092 \
--topic purge-scenario
通过这个方法,kafka会删除主题所有分区的数据。
选择性删除消息
有时可能需要有选择性删除一个或多个主题的分区数据,可以使用kafka-delete-records.sh脚本实现。首先需要在delete-config.json 配置文件中指定分区级偏移量,我们打算分区的删除所有数据,partition指定分区,offset=-1:
{
"partitions": [
{
"topic": "purge-scenario",
"partition": 1,
"offset": -1
}
],
"version": 1
}
接着处理删除记录:
$ bin/kafka-delete-records.sh \
--bootstrap-server localhost:9092 \
--offset-json-file delete-config.json
现在验证从分区0获取数据:
$ bin/kafka-console-consumer.sh \
--bootstrap-server=0.0.0.0:9092 \
--from-beginning --topic purge-scenario --partition=0 \
--max-messages 1 --timeout-ms 1000
44017
Processed a total of 1 messages
接着从分区1获取数据:
$ bin/kafka-console-consumer.sh \
--bootstrap-server=0.0.0.0:9092 \
--from-beginning --topic purge-scenario \
--partition=1 \
--max-messages 1 --timeout-ms 1000
[2021-02-28 11:48:03,548] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.TimeoutException
Processed a total of 0 messages
删除重新创建主题
另外方法是通过删除主题删除其所有数据,然后重新创建主题。当然只有服务端设置delete.topic.enable属性为true才可能删除主题:
$ bin/kafka-server-start.sh config/server.properties \
--override delete.topic.enable=true
可以通过kafka-topics.sh命令删除主题:
$ bin/kafka-topics.sh \
--delete --topic purge-scenario \
--zookeeper localhost:2181
Topic purge-scenario is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
现在验证主题:
$ bin/kafka-topics.sh --zookeeper localhost:2181 --list
如果确认主题已不存在,然后再重新创建主题。
总结
本文介绍了几种方式删除kafka主题数据。包括设置主题过期时间、删除主题所有数据及部分分区数据,到通过删除主题变相删除数据。
版权归原作者 梦想画家 所有, 如有侵权,请联系我们删除。