一、概述

生产环境中,有一个topic的数据量非常大。这些数据不是非常重要,需要定期清理。

要求:默认保持24小时,某些topic 需要保留2小时或者6小时

二、清除方式

主要有3个:

1. 基于时间

2. 基于日志大小

3. 基于日志起始偏移量

详情,请参考链接:

https://blog.csdn.net/u013256816/article/details/80418297

接下来,主要介绍基于时间的清除!

kafka版本为:  2.11-1.1.0

zk版本为:  3.4.13

三、kafka配置

# 启用删除主题
delete.topic.enable=true
# 检查日志段文件的间隔时间,以确定是否文件属性是否到达删除要求。
log.retention.check.interval.ms=

注意:这2行配置必须存在,否则清除策略失效!

log.retention.check.interval.ms 参数的单位是毫秒,这里表示间隔1秒钟

四、清除策略

全局topic

在 server.properties 文件中配置的是全局策略,针对每一个topic

比如:

log.retention.hours=

表示保留3个小时

单个topic

针对单个topic策略,需要使用脚本kafka-configs.sh

此脚本不需要重启kafka就会生效!

首先来查看一下,当前的topic策略,比如test

bin/kafka-configs.sh --zookeeper zookeeper-.default.svc.cluster.local: --describe --entity-type topics --entity-name test

参数解释:

--describe  详细信息

--entity-type 实体类型

--entity-name 指定topic名

输出:

Configs for topic 'test' are

这个表示为策略为空

删除topic数据

如果需要删除topic所有数据,使用命令

bin/kafka-topics.sh --delete --topic test --zookeeper zookeeper-.default.svc.cluster.local:

这个命令,请谨慎执行!!!

如果想保留主题,只删除主题现有数据(log)。可以通过修改数据保留时间实现

bin/kafka-configs.sh --zookeeper zookeeper-.default.svc.cluster.local: --entity-type topics --entity-name test --alter --add-config retention.ms=

执行输出:

Completed Updating config for entity: topic 'test'.

注意:修改保留时间为10秒钟,并不是10秒钟就马上删掉。kafka是采用轮询的方式,轮询到这个topic时,删除10秒钟前的数据。

时间由server.properties里面的log.retention.check.interval.ms选项为主

假设说 log.retention.check.interval.ms 值为1分钟,那么等待70秒,这个topic的数据就会自动被删除!

再次查看topic策略

bin/kafka-configs.sh --zookeeper zookeeper-.default.svc.cluster.local: --describe --entity-type topics --entity-name test

输出:

Configs for topic 'test' are retention.ms=

发现目前的删除策略为 retention.ms=10000

删除策略

如果需要删除上面的10秒策略,使用以下命令:

bin/kafka-configs.sh --zookeeper zookeeper-.default.svc.cluster.local: --entity-type topics --entity-name test --alter --delete-config retention.ms

输出:

Completed Updating config for entity: topic 'test'.

再次查看topic策略

bin/kafka-configs.sh --zookeeper zookeeper-.default.svc.cluster.local: --describe --entity-type topics --entity-name test

输出:

Configs for topic 'test' are

发现策略为空,说明删除成功了!

五、测试清除策略

测试思路

说明:

第一步,设置清除策略为保留10秒

第二步,进入生产者模式,输入消息 a

第三步,等待5秒,再次进入生产者模式,输入消息 b

第四部,进入消费者模式,看输出的消息是a还是b

判断标准:

在进行第三步时,a这条消息,应该已经被删除了。所以在第15秒进入消费者模式时,应该输出 b,这样的话,策略才是成功的!

设置策略

topic 为test的数据保留10秒

bin/kafka-configs.sh --zookeeper zookeeper-.default.svc.cluster.local: --entity-type topics --entity-name test --alter --add-config retention.ms=

生产模式

进入生产模式,输入a

bin/kafka-console-producer.sh --broker-list kafka-1.default.svc.cluster.local:9092 --topic test
> a

等待5秒后,再次进入生产模式,输入b

bin/kafka-console-producer.sh --broker-list kafka-1.default.svc.cluster.local:9092 --topic test
> b

消费者模式

等待5秒后,进入 消费者模式

bin/kafka-console-consumer.sh --bootstrap-server kafka-1.default.svc.cluster.local:9092 --topic test --from-beginning

b

如果消费者输出为b,表示策略成功!

备注:

如果生产环境中,正在不断的进行生产和消费,执行kafka-configs.sh 脚本,是否会有影响呢?

答案是不会的,它是动态策略!

本文参考链接:

https://blog.csdn.net/forrest_ou/article/details/78999983

kafka删除topic数据的更多相关文章

  1. kafka 删除topic清空数据

    原 kafka 删除topic清空数据 2018年11月20日 18:17:50 Ming! 阅读数:1391   版权声明:版权声明:本文为博主原创文章,未经博主允许不得转载. https://bl ...

  2. kafka删除topic及其相关数据

    1.删除kafka存储目录(server.properties文件log.dirs配置,默认为"/tmp/kafka-logs")相关topic目录 2.Kafka 删除topic ...

  3. kafka删除topic的方法及我在kafka上边的一些经验

    我在本地做kafka的producer调试,每隔一段时间后,所使用的topic管道就会堆积数据,而且我这边使用的是  kafka   bin 下的consumer命令单独消费的,每次都是  --fro ...

  4. kafka 清除topic数据脚本

    原 kafka 清除topic数据脚本 2018年07月25日 16:57:13 pete1223 阅读数:1028     #!/bin/sh       param=$1   echo " ...

  5. kafka删除主题数据和删除主题

    1.删除主题 在server.properties中增加设置,默认未开启 delete.topic.enable=true 删除主题命令 /bin/kafka-topics --delete --to ...

  6. kafka删除topic详解

    1.删除topic命令 bin/kafka-topics.sh --delete --zookeeper 192.168.242.131:2181 --topic aaa 注:此命令如果要生效,还需在 ...

  7. kafka 删除topic

    两种方法: 一.修改配置文件server.properties 添加如下配置: delete.topic.enable=true 说明:官方给的文档说明“Enables delete topic. D ...

  8. kafka删除topic

    手动: 删除kafka存储目录(server.properties文件log.dirs配置,默认为"/tmp/kafka-logs")相关topic目录 删除zookeeper & ...

  9. kafka删除topic后再创建同名的topic报错(ERROR org.apache.kafka.common.errors.TopicExistsException)

    [hadoop@datanode3 logs]$ kafka-topics.sh --delete --zookeeper datanode1:2181 --topic firstTopic firs ...

随机推荐

  1. EF6 在 SQLite中使用备忘

    == 菜鸟级选手试验在EF6中使用Sqlite,零EF基础,少量Sqlite基础.经过断断续续的很长时间 - _ -! >>连接 1. 安装 使用目前最新版本EF6.1,Sqlite1.0 ...

  2. codeforces 732D(二分)

    题目链接:http://codeforces.com/contest/732/problem/D 题意:有m门需要过的课程,n天的时间可以选择复习.考试(如果的d[i]为0则只能复习),一门课至少要复 ...

  3. centOS安装nginx

    下载源码 wget http://nginx.org/download/nginx-1.10.1.tar.gz git clone https://bitbucket.org/nginx-goodie ...

  4. UIDatePicker swift

    // // ViewController.swift // UILabelTest // // Created by mac on 15/6/23. // Copyright (c) 2015年 fa ...

  5. (原创)LINUX_UNIX设计思想-读书笔记

    第一章 一.Unit哲学 1.小即是美 2.让每一个程序只做好一件事情 3.尽快建立原型 4.舍高效率而取可移植性 5.使用纯文本文件来存储数据 6.充分利用软件的杠杆效应 7.使用shell脚本来提 ...

  6. javascript字典数据结构Dictionary实现

    <!DOCTYPE html> <html xmlns="http://www.w3.org/1999/xhtml"> <head runat=&qu ...

  7. 前端axios下载excel,并解决axios返回header无法获取所有数据的问题

    需求:通过后端接口下载excel文件,后端没有文件地址,返回二进制流文件 实现:axios(ajax类似) 主要代码: axios:设置返回数据格式为blob或者arraybuffer 如: var ...

  8. java类的加载和执行顺序

    以前面试总会碰到涉及到类的执行过程的笔试题.下面记录我自己的测试结果: public class ClassA { public static ClassA classa = new ClassA() ...

  9. Rabbit and Grass

    链接 [http://acm.hdu.edu.cn/showproblem.php?pid=1849] 题意 大学时光是浪漫的,女生是浪漫的,圣诞更是浪漫的,但是Rabbit和Grass这两个大学女生 ...

  10. Linux中程序开机自启

    Linux中程序开机自启 https://www.itbulu.com/debian-ubuntu-rclocal.html