Kafka日志清除策略

一、更改日志输出级别

config/log4j.properties中日志的级别设置的是TRACE,在长时间运行过程中产生的日志大小吓人,所以如果没有特殊需求,强烈建议将其更改成INFO级别。具体修改方法如下所示,将config/log4j.properties文件中最后的几行中的TRACE改成INFO,修改前如下所示:

log4j.logger.kafka.network.RequestChannel$=TRACE, requestAppender
log4j.additivity.kafka.network.RequestChannel$=false
#log4j.logger.kafka.network.Processor=TRACE, requestAppender
#log4j.logger.kafka.server.KafkaApis=TRACE, requestAppender
#log4j.additivity.kafka.server.KafkaApis=false
log4j.logger.kafka.request.logger=TRACE, requestAppender
log4j.additivity.kafka.request.logger=false
log4j.logger.kafka.controller=TRACE, controllerAppender
log4j.additivity.kafka.controller=false
log4j.logger.state.change.logger=TRACE, stateChangeAppender
log4j.additivity.state.change.logger=false

修改后如下所示:

log4j.logger.kafka.network.RequestChannel$=INFO, requestAppender
log4j.additivity.kafka.network.RequestChannel$=false
#log4j.logger.kafka.network.Processor=INFO, requestAppender
#log4j.logger.kafka.server.KafkaApis=INFO, requestAppender
#log4j.additivity.kafka.server.KafkaApis=false
log4j.logger.kafka.request.logger=INFO, requestAppender
log4j.additivity.kafka.request.logger=false
log4j.logger.kafka.controller=INFO, controllerAppender
log4j.additivity.kafka.controller=false
log4j.logger.state.change.logger=INFO, stateChangeAppender
log4j.additivity.state.change.logger=false

二、利用Kafka日志管理器

Kafka日志管理器允许定制删除策略。目前的策略是删除修改时间在N天之前的日志(按时间删除),也可以使用另外一个策略:保留最后的N GB数据的策略(按大小删除)。为了避免在删除时阻塞读操作,采用了copy-on-write形式的实现,删除操作进行时,读取操作的二分查找功能实际是在一个静态的快照副本上进行的,这类似于Java的CopyOnWriteArrayList。

Kafka消费日志删除思想:Kafka把topic中一个parition大文件分成多个小文件段,通过多个小文件段,就容易定期清除或删除已经消费完文件,减少磁盘占用

log.cleanup.policy=delete启用删除策略
直接删除,删除后的消息不可恢复。可配置以下两个策略:
清理超过指定时间清理:
log.retention.hours=16
超过指定大小后,删除旧的消息:
log.retention.bytes=1073741824

三、压缩策略

将数据压缩,只保留每个key最后一个版本的数据。首先在broker的配置中设置log.cleaner.enable=true启用cleaner,这个默认是关闭的。在Topic的配置中设置log.cleanup.policy=compact启用压缩策略。
  压缩策略的细节如下:

如上图,在整个数据流中,每个Key都有可能出现多次,压缩时将根据Key将消息聚合,只保留最后一次出现时的数据。这样,无论什么时候消费消息,都能拿到每个Key的最新版本的数据。
    压缩后的offset可能是不连续的,比如上图中没有5和7,因为这些offset的消息被merge了,当从这些offset消费消息时,将会拿到比这个offset大的offset对应的消息,比如,当试图获取offset为5的消息时,实际上会拿到offset为6的消息,并从这个位置开始消费。
    这种策略只适合特俗场景,比如消息的key是用户ID,消息体是用户的资料,通过这种压缩策略,整个消息集里就保存了所有用户最新的资料。
    压缩策略支持删除,当某个Key的最新版本的消息没有内容时,这个Key将被删除,这也符合以上逻辑。

时间: 2024-12-15 15:29:09

Kafka日志清除策略的相关文章

关于Kafka日志留存策略的讨论

关于Kafka日志留存(log retention)策略的介绍,网上已有很多文章.不过目前其策略已然发生了一些变化,故本文针对较新版本的Kafka做一次统一的讨论.如果没有显式说明,本文一律以Kafka 1.0.0作为分析对象. 所谓日志留存策略,就是Kafka保存topic数据的规则,我将按照以下几个方面分别介绍留存策略: 留存策略类型 留存机制及其工作原理 一.留存策略类型 目前,与日志留存方式相关的策略类型主要有两种:delete和compact.这两种留存方式的机制完全不同.本文主要讨论

Kafka日志及Topic数据清理

由于项目原因,最近经常碰到Kafka消息队列拥堵的情况.碰到这种情况为了不影响在线系统的正常使用,需要大家手动的清理Kafka Log.但是清理Kafka Log又不能单纯的去删除中间环节产生的日志,中间关联的很多东西需要手动同时去清理,否则可能会导致删除后客户端无法消费的情况. 在介绍手动删除操作之前,先简单的介绍一下Kafka消费Offset原理. 一.Kafka消费Offset 在通过Client端消费Kafka中的消息时,消费的消息会同时在Zookeeper和Kafka Log中保存,如

lagstash + elasticsearch + kibana 3 + kafka 日志管理系统部署 02

因公司数据安全和分析的需要,故调研了一下 GlusterFS + lagstash + elasticsearch + kibana 3 + redis 整合在一起的日志管理应用: 安装,配置过程,使用情况等续 一,glusterfs分布式文件系统部署: 说明: 公司想做网站业务日志及系统日志统一收集和管理,经过对mfs, fastdfs 等分布式文件系统的调研,最后选择了 glusterfs,因为Gluster具有高扩展性.高性能.高可用性.可横向扩展的弹性特点,无元数据服务器设计使glust

日志审计策略配置audit

日志审计策略配置 1.   系统缺省已经开启syslog/rsyslog服务,禁止关闭.系统syslog/rsyslog服务会将所有系统日志自动记录到/var/log/messages文件中,系统日志永久保留. 2.   开启audit审计功能,可以监控指定用户或目录,缺省会监控root的所有登录和操作. l  添加规则到 /etc/audit/audit.rules(RHEL7为/etc/audit/rules.d/audit.rules) 文件中,实现监控所有用户的登录行为,包含用户所有操作

Tomcat 日志清除(含扩展AccessLogValve)

1.tomcat的访问日志AccessLogs定期或者定量删除 a 开启tomcat访问日志 编辑${catalina}/conf/server.xml文件.注:${catalina}是tomcat的安装目录 把以下的注释(<!-- -->)去掉即可. <!--<ValveclassName="org.apache.catalina.valves.AccessLogValve" directory="logs"  prefix="l

Kafka日志报错:[error] k.m.j.KafkaJMX$ - Failed to connect to service:jmx:rmi:///jndi/rmi://10.1.3.116:-1/jmxrmi java.lang.IllegalArgumentException: requirement failed: No jmx port but jmx polling enabled!

一.Kafka日志报错:[error] k.m.j.KafkaJMX$ - Failed to connect to service:jmx:rmi:///jndi/rmi://10.1.3.116:-1/jmxrmi java.lang.IllegalArgumentException: requirement failed: No jmx port but jmx polling enabled! 以上报错,并没有对Kafka对使用造成直接影响: 1>.启动生产者:kafka-console

kafka 日志结构

1.kafka日志结构 直接举例子: 例如kafka有个名字叫 haha 的topic,那么kafka日志下面有kafka-0,kafka-1,kafka-2...,kafka-n,具体多少个,创建分区的时候有几个分区,n就是几. [[email protected] kafka]# ls haha- haha-0/ haha-1/ haha-2/ 每个目录下面有三个日志文件,xxx.log是消息集文件, xxx.index 偏移量索引文件 ,xxx.timeindex 时间戳索引文件 [[em

性能优化分析Spring Cloud ELK+kafka日志分析平台

一.概述 在笔者的上一篇博客介绍了Spring Cloud ELK+kafka日志分析平台的搭建,http://xuyangyang.club/articles/2018/05/24/1527176074152.html,但是笔者在测试环境中发现,在logstash采用了grok插件去处理日志埋点和解析的时候发现了高资源占用,在阿里云8核16G的服务器部署后,测试环境大概每秒不超过几百条的日志的解析下竟然CPU占用高达95%左右,笔者分析了其中的原因,首先由于几个服务的日志格式相关配置还没有落地

asp.net生产环境和开发环境的错误日志包装策略

对于错误日志的输出,我们借助web.config的两个标志位: <!--全局包装异常处理页面,只有在PageError和Application_Error做清除错误操作才可不跳转--> <!--调试时请设置成Off,生产环境此项必须为On--> <customErrors mode="Off" defaultRedirect="~/ErrorPage/404/index.html"> <error statusCode=&q