思想
Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。清理策略针对的是过期的segment文件，而不是某条过期的数据。可以单独针对某topic配置，也可针对kafka集群配置（config/server.properties）。策略分三种：基于时间，基于日志文件大小，基于日志文件起始偏移量。为了避免在删除时阻塞读操作，采用了copy-on-write形式的实现，删除操作进行时，读取操作的二分查找功能实际是在一个静态的快照副本上进行的，这类似于Java的CopyOnWriteArrayList。
topic级别修改命令
修改单个topic清理策略：bin/kafka-configs.sh --zookeeper ZK_IP:2181 --describe --entity-type topics --entity-name test
删除topic：bin/kafka-topics.sh --delete --topic test --zookeeper ZK_IP:2181
保留topic，删除数据，通过修改数据保留时间实现（如下：保留10s）：bin/kafka-configs.sh --zookeeper ZK_IP:2181 --entity-type topics --entity-name test --alter --add-config retention.ms=10000
查看topic清理策略：bin/kafka-configs.sh --zookeeper ZK_IP:2181 --describe --entity-type topics --entity-name test
删除策略：bin/kafka-configs.sh --zookeeper ZK_IP:2181 --entity-type topics --entity-name test --alter --delete-config retention.ms
配置文件级别删除
通过log.cleanup.policy，分为删除和压缩。
删除策略
当日志达到log.segment.bytes大小，会创建新的segment。当segment超过log.segment.bytes或保留时长达到log.retention.hours，就会被清理掉。
| 属性名 | 含义 | 默认值 |
|---|
| log.cleanup.polict | 日志清理保存策略，delete/compact | delete | | log.retention.hours | 日志保存时间，可以选择hours,minutes和ms | 168(7day) | | log.retention.bytes | 删除前日志文件允许保存的最大值 | -1 | | log.segment.delete.delay.ms | 日志文件被真正删除前的保留时间 | 60000(1min) | | log.cleanup.interval.mins | 清理工作执行时间间隔 | 10 | | log.retention.check.interval.ms | 清理工作执行时间间隔（新版本） | 300000 |
达到清理条件的日志文件，进行“delete”标注，文件无法被索引到。log.segment.delete.delay.ms这个时间后，文件才会被真正的从文件系统中删除。
配置server.properties，添加删除策略：
# 默认delete策略
# 日志保留时间
log.retention.hours=168
# 超过此大小的segment被删除，默认1G
#log.retention.bytes=1073741824
# 达到此大小，创建一个新的segment，默认1G
log.segment.bytes=1073741824
# 清理工作执行的间隔
log.retention.check.interval.ms=300000
刷新策略
| 属性名 | 含义 | 默认值 |
|---|
| log.flush.interval.messages | 消息达到该条时，将数据写入到日志文件 | 10000 | | log.flush.interval.ms | 当达到该时间间隔时，强制执行一次flush | null | | log.flush.scheduler.interval.ms | 周期性检查，是否需要将消息flush | - |
分段策略属性
| 属性名 | 含义 | 默认值 |
|---|
| log.roll.{hours,ms} | 日志滚动的周期时间，到达指定周期时间时，强制生成一个新的segment | 168（7day） | | log.segment.bytes | 每个segment的最大容量。到达指定容量时，将强制生成一个新的segment | 1G(-1表示不限制) | | log.retention.check.interval.ms | 日志片段文件检查的周期时间 | 60000 |
压缩
将数据压缩，只保留每个key最后一个版本的数据，offset可能不连续。
#启用cleaner，默认是关闭的
log.cleaner.enable=true
log.cleanup.policy=compact
这种策略只适合特殊场景，比如消息的key是用户ID，消息体是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。压缩策略支持删除，当某个Key的最新版本的消息没有内容时，这个Key将被删除，这也符合以上逻辑。
据起始偏移量删除
日志配置
修改日志级别为INFO，防止长时间产生过多日志。将config/log4j.properties中必要选项改为INFO。
修改日志（server.log, controller.log, state-change.log等）存放位置，放置长时间产生日志占满磁盘。可以将日志目录配置到更大空间的分区盘。可通过更改kafka-run-class.sh中的LOG_DIR变量值实现。
参考
https://blog.csdn.net/shykevin/article/details/90103364.
http://www.openskill.cn/article/550.
https://blog.csdn.net/chenyixin121738/article/details/109590355.
https://blog.csdn.net/u013256816/article/details/80418297.
|