Apache Kafka中数据清理机制涉及的主要概念有两个：基于时间或大小的日志保留策略，以及日志压缩。这些特性允许Kafka管理其存储空间，保留有用的数据，同时清除过时或重复的数据。
基于时间或大小的日志保留策略
日志保留策略配置参数：
log.retention.hours：日志保留的小时数。log.retention.bytes：日志保留的最大字节数。
Kafka的日志保留策略是通过删除旧的日志段来实现的。日志段（Log Segment）是Kafka日志的分割单位。
源码解析
在Kafka的源码中，日志保留策略主要是通过LogManager类管理的，具体实现在Log类的deleteOldSegments方法中。
// kafka/log/Log.scalaprivate def deleteOldSegments(predicate: (LogSegment) => Boolean): Int = {// ...中间代码省略...val lastClean = activeSegment.baseOffset // the active segment is always keptval deletable = logSegments.view.takeWhile(_.baseOffset < lastClean)val numToDelete = deletable.count(predicate)// ...中间代码省略...if(numToDelete > 0) {// Actually delete the segments marked for deletionlogSegments.view.take(numToDelete).foreach(deleteSegment)}numToDelete
}private def deleteSegment(segment: LogSegment) {// ...中间代码省略...segment.delete()// ...中间代码省略...
}
代码示例
# 设置主题的日志保留时间为2天
kafka-configs --zookeeper localhost:2181 --entity-type topics --entity-name your_topic --alter --add-config retention.ms=172800000# 设置主题的日志保畛大小为1GB
kafka-configs --zookeeper localhost:2181 --entity-type topics --entity-name your_topic --alter --add-config retention.bytes=1073741824
日志压缩
日志压缩允许Kafka保留至少一条每个键的最新消息，同时删除任何旧的、重复的键值记录。
配置参数
cleanup.policy=compact：启用日志压缩。min.cleanable.dirty.ratio：控制启动压缩任务前允许的最大脏数据比例。
源码解析
日志压缩在LogCleaner类中实现，它由一个或多个CleanerThread组成，不断循环遍历所有日志。
// kafka/log/LogCleaner.scalaclass LogCleaner(config: CleanerConfig, logDirs: Array[File], logs: Pool[TopicAndPartition, Log], time: Time) extends Logging {// ...中间代码省略...private val cleaners = (0 until config.numThreads).map(i => new CleanerThread(i, logs, cleanerConfig, time))// ...中间代码省略...
}class CleanerThread(threadId: Int, logs: Pool[TopicAndPartition, Log], cleanerConfig: CleanerConfig, time: Time) extends ShutdownableThread {// ...中间代码省略...override def doWork() {// ...中间代码省略...cleanOrSleep()// ...中间代码省略...}// ...中间代码省略...
}
代码示例
# 创建带有日志压缩策略的主题
kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic compacted_topic --config cleanup.policy=compact# 更改现有主题的配置以启用日志压缩
kafka-configs --zookeeper localhost:2181 --entity-type topics --entity-name your_existing_topic --alter --add-config cleanup.policy=compact
注意事项
- Kafka的日志清理操作是异步的，不会立即反映更改。
- 对于压缩，键值是必须的，没有键的消息将不会被压缩。
- 在压缩过程中，CPU和IO可能会有显著使用，这可能影响到Kafka的性能。
- Kafka的日志清理与日志追加操作是并发进行的，设计上保证了数据的一致性。
Kafka源码提供了对数据清理机制的更深入理解。如果想要更详细地探索，可以直接查阅Kafka源码。
)
