Kafka delete 日志清理策略(日志删除)
- 原理:按照一定保留策略,直接删除不符合条件的日志分段。Kafka 把 topic 的一个 partition 大文件分成多个小文件段,通过这种方式,能方便地定期清除或删除已消费完的文件,以减少磁盘占用 。
- 保留策略
- 按时间删除:设定一个时间阈值,删除修改时间在该时间之前的日志。比如设置
log.retention.hours = 1
,就表示只保存 1 小时内的日志,超出 1 小时的日志分段会被删除。 - 按大小删除:指定一个数据大小阈值,当日志数据超过这个大小时,执行删除操作,保留最后的指定大小数据。例如
log.retention.bytes = 1073741824
(即 1GB ),表示日志数据超过 1GB 时,会删除旧的消息。
- 相关参数
log.cleanup.policy = delete
:启用删除策略。log.retention.check.interval.ms
:专门的日志删除任务周期性检测的时间间隔,默认 300000ms(5 分钟 ),即每隔 5 分钟检查一次是否有符合删除条件的日志分段 。log.retention.hours
:按时间清理时,指定日志保留的小时数 。log.retention.bytes
:按大小清理时,指定日志保留的字节数 。需注意log.retention.bytes
和log.retention.hours
任意一个达到要求,都会执行删除操作,且会被 topic 创建时的指定参数覆盖 。
Kafka compact 日志清理策略(日志压缩)
- 原理:针对每个消息的 key 进行整合,对于有相同 key 的不同 value 值,只保留最后一个版本 。就像在一个记录集合里,相同标识(key )的记录,只留下最新的那条 。比如一个 key 对应的值先后为 “值 1”“值 2”“值 3” ,经过日志压缩后,只会保留 “值 3” 。清理重复 key 后,一些 segment 文件大小会变小,Kafka 会将小文件再合并成大的 segment 文件 。
- 特殊处理:当某个 key 的最新版本消息没有内容(value 为 null )时,这个 key 将被删除,这类消息被称为 “墓碑消息(tombstone )” 。墓碑消息的存放时间和 broker 的配置
log.cleaner.delete.retention.ms
有关,默认值是 24 小时 。在执行日志清理时,会删除到期的墓碑消息 。 - 清理流程
- 对于每个 Kafka partition 的日志,以 segment 为单位,分为已清理和未清理部分,未清理部分又细分为可清理和不可清理的。
- 每个日志目录下有
cleaner - offset - checkpoint
文件记录清理进度。 - 找出可清理的 segment(active segment 不能清理,根据
min.compaction.lag.ms
配置判断其他 segment 是否能清理 ,即判断 segment 最后一条记录的插入时间是否超过最小保留时间 )。 - 构建 SkimpyOffsetMap 对象(key 与 offset 的映射哈希表 ),遍历可清理 segment 的每条日志,将 key 和 offset 存入其中。
- 再遍历已清理和可清理部分的 segment 日志,根据 SkimpyOffsetMap 判断是否保留日志 。
- 执行清理操作,将可清理部分的 segment 变为已清理的,同时更新 cleaner checkpoint 记录的 offset 。
- 相关参数
log.cleaner.enable = true
:开启日志压缩功能。log.cleanup.policy = compact
:启用日志压缩策略 。min.compaction.lag.ms
:设置日志段中消息可被压缩的最小时间间隔 ,确保消息存留一定时间后才可能被清理 。log.cleaner.delete.retention.ms
:墓碑消息的保留时长 。