大数据-70 Kafka 高级特性 物理存储 日志存储 日志清理: 日志删除与日志压缩_kafka logs文件清理

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

• Hadoop（已更完）
• HDFS（已更完）
• MapReduce（已更完）
• Hive（已更完）
• Flume（已更完）
• Sqoop（已更完）
• Zookeeper（已更完）
• HBase（已更完）
• Redis （已更完）
• Kafka（正在更新…）

章节内容

上节我们完成了如下内容：

• 日志索引文件
• 查看物理存储、查看详细的索引文件
• 消息偏移
• 偏移量的存储

日志清理

Kafka提供两种日志清理策略：

• 日志删除：按照一定的删除策略，将不满足条件的数据进行数据删除
• 日志压缩：针对每个消息的Key进行整合，对于有相同的Key的不同Value值，只保留最后一个版本。
• Kafka提供 log.cleanup.policy 参数进行相应设置，默认值：delete。还可以选择compact。
• 主题级别的配置是：cleanup.policy

日志删除

基于时间

日志删除任务会根据 log.retention.hours / log.retention.minutes / log.retention.ms 设定日志保留的时间节点。如果超过该设定值，就需要进行删除。默认是7天，log.retention.ms 优先级最高。

Kafka依据日志分段中最大的时间戳进行定位。首先要查询日志分段所对应的时间戳文件，查找时间索引文件中最后一个索引项，若最后一条索引项的时间戳字段大于0，则取该值，否则取最近修改时间。

为什么不直接选最近修改时间？

因为日志文件可以有意无意的被修改，并不能真实的反应日志分段的最大时间消息。

删除过程

• 从日志对象中所维护日志分段的跳跃表中移除待删除的日志分段，保证没有现成对这些日志分段进行读取操作。
• 这些日志分段上所有文件添加上 .delete 后缀。
• 交由一个 delete-file 命名的延迟任务来删除这些 .delete 为后缀的文件，延迟执行时间可以通过 file.delete.delay.ms 进行设置。

如果活跃日志分段中存在需要删除的数据？

• Kafka会切分出一个新的日志分段作为活跃的日志分段，该日志分段不删除，删除原来的日志分段。
• 先腾出地方，再删除。

基于日志大小

日志删除任务会检查当前日志的大小是否超过设定值，设定项为：log.retention.bytes。单个日志分段的大小由 log.segement.bytes 进行设定。

删除过程

• 计算需要被删除的日志总大小（当前日志大小（所有分段）减去retention值）
• 从日志文件第一个LogSegment开始查找可删除的日志分段的文件集合
• 执行删除

基于偏移量

根据日志分段的下一个日志分段的起始偏移量是否大于等于日志文件的起始偏移量，若是，则可以删除日志分段。

删除过程

• 从头开始遍历每个日志分段，日志分段1的下一个日志分段的起始偏移量为21，小于LogStartOffset，将日志分段1加入到删除队列中
• 日志分段2的下一个日志分段的起始偏移量35，小于LogStartOffset，将日志分段2加入到删除队列中
• 日志分段3的下一个日志分段的起始偏移量57，小于LogStartOffset，将日志分段3加入到删除队列中
• 日志分段4的下一个日志分段的起始偏移量71，大于LogStartOffset，则不进行删除。

日志压缩

基础概念

日志压缩是Kafka的一种机制，可以提供较为细粒度的记录保留，而不是基于粗粒度的基于时间保留。
对于具有相同Key，而数据不同，只保留最后一条数据，前面的数据在适合情况下删除。

应用场景

日志压缩特性，就实时计算来说，可以在异常容灾方面有很好的用途。
比如，我们在Spark、Flink中做实时计算时，需要在内存中维护一些数据，这些数据可能是通过聚合了一天或者一周的日志得到的，这些数据一旦由于异常（内存、网络、硬盘）崩溃了，从头开始计算是需要很长时间的。

一个比较可行的方案就是定时将内存里的数据备份到外部存储介质中，当出现崩溃时，再从外部存介质中恢复并继续计算。

使用日志压缩来替代这些外部存储有哪些优势和好处？

• Kafka即是数据源又是存储工具，可以简化技术栈，降低维护成本
• 使用外部存储介质的话，需要将存储的Key记录下来，恢复的时候再使用这些Key将数据取回，实现起来有一定的工程难度和复杂度。使用Kafka的日志压缩特性，只需要把数据写入Kafka，等异常出现恢复任务再读回内存就可以了
• Kafka对于磁盘的读写做了大量的优化工作，比如磁盘顺序读写。相对于外部存储介质没有索引查询等工作量负担，可以实现高性能。同时，Kafka的日志压缩机制可以充分利用廉价的磁盘，不用依赖昂贵的内存来处理，在性能相似的情况下，实现非常高的性价比（仅针对异常处理和容灾的场景）。

日志压缩实现细节

主题的 cleanup.policy 需要设置为：compact
Kafka后台线程会定时将Topic遍历两次：

• 记录每个Key的Hash值最后一次出现的偏移量
• 第二次检查每个Offset对应的Key是否在后面的日志中出现过，如果出现了就删除对应的日志。

日志压缩允许删除，除最后一个key外，删除先前出现的所有该Key对应的记录，在一段时间后从日志中清理以释放空间。
注意：日志压缩与Key有关，确保每个消息的Key不为Null。

压缩是在Kafka后台通过定时重新打开Segment来完成的，Segment压缩细节如下图所示：

日志压缩可以确保：

• 任何保持在日志头部以内的使用者都将看到所写的每条消息，这些消息将具有顺序偏移量。
• 可以使用Topic的min.compation.lag.ms属性来保证消息在被压缩之前必须经过的最短时间，也就是说，它为每个消息（未压缩）头部停留的时间提供下一个下限。可以使用Topic的max.compactiton.lag.ms属性来保证从收到消息符合压缩条件之间的最大延时
• 消息始终保证顺序，压缩永远不会重新排序消息，只是删除一些而已
• 消息的偏移量永远不会改变，它是日志中位置的永久标识
• 从日志开始的任何使用者将至少看到所有记录的最终状态，按记录的顺序写入。
• 另外，如果使用者在比Topic的log.cleaner.delete.retention.ms短的时间内到达日志的头部，则会看到已删除的所有的delete标记， 保留时间默认是24小时。

默认情况下，启动日志清理器，若需要启动特定Topic的日志清理，请添加特定的属性。

日志清理器配置

配置日志清理器，这里为大家总结了以下几点：

• log.cleanup.policy设置compact，Broker的配置，影响集群中所有Topic
• log.cleaner.min.compaction.lag.ms，用于防止对更新超过最小信息进行压缩，如果没有设置，除最后一个Segment之外，所有Segment都有资格进行压缩。

Kafka的日志压缩原理并不复杂，就是定时把所有的日志读取两遍，写一遍，而CPU的速度超过磁盘不是问题，只要日志的量对应的读取两遍和写入一遍的时间在接受范围内，那么它的性能就是可以接受的。