本文我们探讨几种关于如何删除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主题数据。包括设置主题过期时间、删除主题所有数据及部分分区数据,到通过删除主题变相删除数据。

Logo

为开发者提供学习成长、分享交流、生态实践、资源工具等服务,帮助开发者快速成长。

更多推荐