欢迎支持笔者新作:《深入理解Kafka:核心设计与实践原理》和《RabbitMQ实战指南》,同时欢迎关注笔者的微信公众号:朱小厮的博客。

欢迎跳转到本文的原文链接:https://honeypps.com/mq/log-deletion-of-kafka-log-retention/


Kafka将消息存储在磁盘中,为了控制磁盘占用空间的不断增加就需要对消息做一定的清理操作。Kafka中每一个分区partition都对应一个日志文件,而日志文件又可以分为多个日志分段文件,这样也便于日志的清理操作。Kafka提供了两种日志清理策略:

  1. 日志删除(Log Deletion):按照一定的保留策略来直接删除不符合条件的日志分段。
  2. 日志压缩(Log Compaction):针对每个消息的key进行整合,对于有相同key的的不同value值,只保留最后一个版本。

我们可以通过broker端参数log.cleanup.policy来设置日志清理策略,此参数默认值为“delete”,即采用日志删除的清理策略。如果要采用日志压缩的清理策略的话,就需要将log.cleanup.policy设置为“compact”,并且还需要将log.cleaner.enable(默认值为true)设定为true。通过将log.cleanup.policy参数设置为“delete,compact”还可以同时支持日志删除和日志压缩两种策略。日志清理的粒度可以控制到topic级别,比如与log.cleanup.policy对应的主题级别的参数为cleanup.policy,为了简化说明,本文只采用broker端参数做陈述,如若需要topic级别的参数可以查看官方文档。

日志删除

Kafka日志管理器中会有一个专门的日志删除任务来周期性检测和删除不符合保留条件的日志分段文件,这个周期可以通过broker端参数log.retention.check.interval.ms来配置,默认值为300,000,即5分钟。当前日志分段的保留策略有3种:基于时间的保留策略、基于日志大小的保留策略以及基于日志起始偏移量的保留策略。

1. 基于时间

日志删除任务会检查当前日志文件中是否有保留时间超过设定的阈值retentionMs来寻找可删除的的日志分段文件集合deletableSegments,参考下图所示。retentionMs可以通过broker端参数log.retention.hours、log.retention.minutes以及log.retention.ms来配置,其中log.retention.ms的优先级最高,log.retention.minutes次之,log.retention.hours最低。默认情况下只配置了log.retention.hours参数,其值为168,故默认情况下日志分段文件的保留时间为7天。

查找过期的日志分段文件,并不是简单地根据日志分段的最近修改时间lastModifiedTime来计算,而是根据日志分段中最大的时间戳largestTimeStamp来计算。因为日志分段的lastModifiedTime可以被有意或者无意的修改,比如执行了touch操作,或者分区副本进行了重新分配,lastModifiedTime并不能真实地反映出日志分段在磁盘的保留时间。要获取日志分段中的最大时间戳largestTimeStamp的值,首先要查询该日志分段所对应的时间戳索引文件,查找时间戳索引文件中最后一条索引项,若最后一条索引项的时间戳字段值大于0,则取其值,否则才设置为最近修改时间lastModifiedTime。

若待删除的日志分段的总数等于该日志文件中所有的日志分段的数量,那么说明所有的日志分段都已过期,但是该日志文件中还要有一个日志分段来用于接收消息的写入,即必须要保证有一个活跃的日志分段activeSegment,在此种情况下,会先切分出一个新的日志分段作为activeSegment,然后再执行删除操作。

删除日志分段时,首先会从日志文件对象中所维护日志分段的跳跃表中移除待删除的日志分段,以保证没有线程对这些日志分段进行读取操作。然后将日志分段文件添加上“.deleted”的后缀,当然也包括日志分段对应的索引文件。最后交由一个以“delete-file”命名的延迟任务来删除这些“.deleted”为后缀的文件,这个任务的延迟执行时间可以通过file.delete.delay.ms参数来设置,默认值为60000,即1分钟。

2. 基于日志大小

日志删除任务会检查当前日志的大小是否超过设定的阈值retentionSize来寻找可删除的日志分段的文件集合deletableSegments,参考下图所示。retentionSize可以通过broker端参数log.retention.bytes来配置,默认值为-1,表示无穷大。注意log.retention.bytes配置的是日志文件的总大小,而不是单个的日志分段的大小,一个日志文件包含多个日志分段。

基于日志大小的保留策略与基于时间的保留策略类似,其首先计算日志文件的总大小size和retentionSize的差值diff,即计算需要删除的日志总大小,然后从日志文件中的第一个日志分段开始进行查找可删除的日志分段的文件集合deletableSegments。查找出deletableSegments之后就执行删除操作,这个删除操作和基于时间的保留策略的删除操作相同,这里不再赘述。

3. 基于日志起始偏移量

一般情况下日志文件的起始偏移量logStartOffset等于第一个日志分段的baseOffset,但是这并不是绝对的,logStartOffset的值可以通过DeleteRecordsRequest请求、日志的清理和截断等操作修改。

基于日志起始偏移量的删除策略的判断依据是某日志分段的下一个日志分段的起始偏移量baseOffset是否小于等于logStartOffset,若是则可以删除此日志分段。参考上图,假设logStartOffset等于25,日志分段1的起始偏移量为0,日志分段2的起始偏移量为11,日志分段3的起始偏移为23,那么我们通过如下动作收集可删除的日志分段的文件集合deletableSegments:

  1. 从头开始遍历每个日志分段,日志分段1的下一个日志分段的起始偏移量为11,小于logStartOffset的大小,将日志分段1加入到deletableSegments中;
  2. 日志分段2的下一个日志偏移量的起始偏移量为23,也小于logStartOffset的大小,将日志分段2页加入到deletableSegments中;
  3. 日志分段3的下一个日志偏移量在logStartOffset的右侧,故从日志分段3开始的所有日志分段都不会被加入到deletableSegments中。

收集完可删除的日志分段的文件集合之后的删除操作同基于日志大小的保留策略和基于时间的保留策略相同,这里不再赘述。

欢迎跳转到本文的原文链接:https://honeypps.com/mq/log-deletion-of-kafka-log-retention/


欢迎支持笔者新作:《深入理解Kafka:核心设计与实践原理》和《RabbitMQ实战指南》,同时欢迎关注笔者的微信公众号:朱小厮的博客。


Kafka日志清理之Log Deletion相关推荐

  1. Kafka日志清理之Log Compaction

    欢迎支持笔者新作:<深入理解Kafka:核心设计与实践原理>和<RabbitMQ实战指南>,同时欢迎关注笔者的微信公众号:朱小厮的博客. 欢迎跳转到本文的原文链接:https: ...

  2. Kafka中数据清理(Log Deletion)

    目录 5. Kafka中数据清理(Log Deletion) 5.1 日志删除 5.1.1 定时日志删除任务 5.1.2 基于时间的保留策略 5.1.3 基于日志大小的保留策略 5.1.4 基于日志起 ...

  3. kafka日志清理策略,compact和delete

    文章目录 1. kafka日志清理策略概述 2. kafka segment 2.1 segmnet 的作用 2.2 segment生成相关的配置 3. 日志清理delete策略 3.1 delete ...

  4. 清理offset_关于 kafka 日志清理策略的问题

    现象: 搭建了一个 kafka 服务, 使用 kafka-python 包正常生产数据, 但是 kafka 过五分钟就把我的 topic 删除掉. 但是配置 log 的已经配置了, 我认为 kafka ...

  5. 你花了多久弄明白架构设计?kafka日志清理

    大数据.算法项目在任何大厂无论是面试还是工作运用都是非常广泛的,我们精选了50个百度.腾讯.阿里等大厂的大数据.算法落地经验甩给大家,千万不要做收藏党哦,空闲时间记得随时看看! 如果你没有大厂项目经验 ...

  6. kafka 日志相关配置

    日志目录 ${kafka.logs.dir}/server.log :服务器日志 ${kafka.logs.dir}/state-change.log:状态变化日志 ${kafka.logs.dir} ...

  7. Kafka日志模块(九):LogCleaner(重复日志数据清理compact)

    之前介绍了按照时间空间老化消息的定时任务,本篇来看一下 LogCleaner 线程,如果在配置中指定了 log.cleaner.enable=true,那么在 LogManager#startup 方 ...

  8. CC00065.kafka——|Hadoopkafka.V50|——|kafka.v50|日志清理|

    一.日志压缩策略 ### --- 概念~~~ 日志压缩是Kafka的一种机制,可以提供较为细粒度的记录保留, ~~~ 而不是基于粗粒度的基于时间的保留. ~~~ 对于具有相同的Key,而数据不同,只保 ...

  9. 学习Linux系统清理/var/log/journal/垃圾日志教程

    学习Linux系统清理/var/log/journal/垃圾日志教程 journal介绍 常见的日志文件 查看垃圾文件的方法 清空 /var/log/journal 文件的方法 1.用 echo 命令 ...

最新文章

  1. java sha1加密ascii码_请问下面java的Sha1加密在c#中对应要怎么写?
  2. java反序列化 exp_java反序列化-ysoserial-调试分析总结篇(4)
  3. 多线程处理器 适用于 已知要处理任务的个数,进行多线程处理
  4. MySQL数据库操作(DDL)
  5. wireshark提取流量包中的文件_从Wireshark监听的数据中提取需要的数据
  6. HTML5需要学html4吗,html5和html4的区别是什么
  7. 6月26号.NET面试题(程序题部分)只要做懂这3道题肯定能脱离菜鸟称号!
  8. python初级进阶篇
  9. OSPF——通告静态缺省(默认)路由(含配置)详解
  10. ningx修改mysql数据库密码_windows下面的php+mysql+nginx
  11. 教育培训机构管理系统软件如何查看优惠券记录?
  12. 四川锦城学院计算机专业好不,四川大学锦城学院计算机专业如何?
  13. AR大屏互动的原理是什么?可以应用在哪些场景中?
  14. 蚂蚁市场教程 电视上怎么安装看快手TV版
  15. 全志A31下5M的CMOSCamera移植修改记录表
  16. vm虚拟机装2019最新kali系统
  17. 文件夹下的多个sdf文件批量转化为csv
  18. jzoj 3234. 阴阳
  19. 数据挖掘-序列模式挖掘-prefixspan算法(样例)
  20. 读王垠的“清华梦的粉碎”有感

热门文章

  1. 【报错】ValueError: not enough values to unpack expected 2, got1
  2. python的md5
  3. 贴片电容耐压值一般都是多少?
  4. 洛谷P2566 [SCOI2009]围豆豆(状压dp+计算几何)
  5. 针对不同创业阶段的创业者适合参加哪些创业赛事活动呢?创业是一场异常艰辛的马拉松,坚持是唯一的捷径,借...
  6. Java中String、StringBuffer、StringBuilder的区别
  7. spring MVC 工作原理
  8. CodeForces - 1267A Apprentice Learning Trajectory(贪心)
  9. HDU - 4552 怪盗基德的挑战书(后缀数组+RMQ/KMP+dp)
  10. CodeForces - 1284B New Year and Ascent Sequence(乱搞)