IT评测·应用市场-qidao123.com

标题: Kafka磁盘写满日记清理操纵 [打印本页]

---

作者: 自由的羽毛    时间: 2024-7-27 05:53
标题: Kafka磁盘写满日记清理操纵
最近项目组的kafka集群，总是由于应用端写入kafka topic的消息太多，导致所在的broker节点占满，导致其他的组件接连宕机。
这里和应用端沟通可以删除1天之前的消息来清理磁盘，并且可以调解topic的消息存活时间。
一、调解Topic的消息存活时长删除消息

1. kafka-configs --zookeeper localhost:2181 --entity-type topics --entity-name topicName --alter --add-config retention.ms=86400000

复制代码

如上调解topic的消息存活时长为为1天，当执行完之后执行查询topic详细信息，可以看到已经发生了修改，并且过一会过期的消息会被删除。

1. kafka-topics --bootstrap-server localhost:9092 --describe --topic topicName

复制代码

二、不修改Topic消息存活时长删除消息

1.登录到相应的呆板上。
2.找到写满的磁盘，删撤除不需要的业务数据。数据清理原则：

• 不可直接删除Kafka的数据目录，避免造成不须要的数据丢失。
  
• 找到占用空间较多或者明白不需要的Topic，选择其中某些Partition，从最早的日记数据开始删除。删除segment及相应地index和timeindex文件。不要清理内置的Topic，例如__consumer_offsets和_schema等。
  

3.重启磁盘被写满的相应的Broker节点，使日记目录online。
参考：Kafka磁盘写满时怎样运维操纵_开源大数据平台E-MapReduce-阿里云帮助中心 (aliyun.com)
怎么删除kafka中的数据-火山引擎 (volcengine.com)

三、Kafka消息清理策略

在Kafka中，存在数据过期的机制，称为data expire。怎样处置惩罚过期数据是根据指定的policy（策略）决定的，而处置惩罚过期数据的活动，即为log cleanup。
在Kafka中有以下几种处置惩罚过期数据的策略：
log.cleanup.policy=delete（Kafka中所有用户创建的topics，默认均为此策略）

• 根据数据已保存的时间，举行删除（默以为1周）
  
• 根据log的max size，举行删除（默以为-1，也就是无穷制）
  

log.cleanup.policy=compact（topic __consumer_offsets 默以为此策略）

• 根据messages中的key，举行删除操纵
  
• 在active segment 被commit 后，会删撤除old duplicate keys
  
• 无穷制的时间与空间的日记保留
  

主动清理Kafka中的数据可以控制磁盘上数据的大小、删除不需要的数据，同时也淘汰了对Kafka集群的维护成本。
那Log cleanup 在什么时候发生呢？

• 起首值得注意的是：log cleanup 在partition segment 上发生
  
• 更小/更多的segment，也就意味着log cleanup 发生的频率会上升
  
• Log cleanup 不应该频繁发生=> 因为它会斲丧CPU与内存资源
  
• Cleaner的查抄会在每15秒举行一次，由log.cleaner.backoff.ms 控制
  

log.cleanup.policy=delete

log.cleanup.policy=delete 的策略，根据数据保留的时间、以及log的max size，对数据举行cleanup。控制数据保留时间以及log max size的参数分别为：
log.retention.hours：指定数据保留的时常（默以为一周，168）

• 将参数调解到更高的值，也就意味着会占据更多的磁盘空间
  
• 更小值意味着保存的数据量会更少（如果consumer 宕机超过一周，则数据便会再未处置惩罚前即丢失）
  

log.retention.bytes：每个partition中保存的最大数据量大小（默以为-1，也就是无穷大）

• 再控制log的大小不超过一个阈值时，会比较有用
  

在到达log cleanup 的条件后，cleaner会主动根据时间或是空间的规则举行删除，新数据仍写入active segment：

针对于这个参数，一般有以下两种使用场景，分别为：
log保留周期为一周，根据log保留期举行log cleanup：

• log.retention.hours=168 以及 log.retention.bytes=-1
  

log保留期为无穷制，根据log大小举行举行log cleanup：

• log.retention.hours=17520以及 log.retention.bytes=524288000
  

其中第一个场景会更常见。

Log Compaction

Log compaction用于确保：在一个partition中，对恣意一个key，它所对应的value都是最新的。
这里举个例子：我们有个topic名为employee-salary，我们希望维护每个employee当前最新的工资情况。
左边的是compaction前，segments中的数据，右边为compaction 后，segments中的数据，其中有部门key对应的value有更新：

  
可以看到在log compaction后，相对于更新后的key-value message，旧的message被删除。
Log Compaction 有如下特点：

• messages的顺序仍旧是保留的，log compaction 仅移除一些messages，但不会重新对它们举行排序
  
• 一条message的offset是无法改变的（immutable），如果一条message缺失，则offset会直接被跳过
  
• 被删除的records在一段时间内仍旧可以被consumers访问到，这段时间由参数delete.retention.ms（默以为24小时）控制
  

需要注意的是：Kafka 本身是不会构造用户发送duplicate data的。这些重复数据也仅会在一个segment在被commit 的时候做重复数据删除，所以consumer仍会读取到这部门重复数据（如果客户端有发的话）。
Log Compaction也会有时失败，compaction thread 可能会crash，所以需要确保给Kafka server 足够的内存用于做这些操纵。如果log compaction异常，则需要重启Kafka（此为一个已知的bug）。
Log Compaction也无法通过API手动触发（至少到现在为止是这样），只能server端主动触发。
下面是一个 Log Compaction过程的示意图：

正在写入的records仍会被写入Active Segment，已经committed segments会主动做compaction。此过程会遍历所有segments中的records，并移撤除所有需要被移除的messages。
Log compaction由上文提到的log.cleanup.policy=compact举行配置，其中：

• Segment.ms（默以为7天）：在关闭一个active segment前，所需等候的最长时间
  
• Segment.bytes（默以为1G）：一个segment的最大大小
  
• Min.compaction .lag.ms（默以为0）：在一个message可以被compact前，所需等候的时间
  
• Delete.retention.ms（默以为24小时）：在一条message被加上删除标志后，在实际删除前等候的时间
  
• Min.Cleanable.dirty.ratio（默以为0.5）：若是设置的更高，则会有更高效的清理，但是更少的清理操纵触发。若是设置的更低，则清理的效率稍低，但是会有更多的清理操纵被触发
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/)

Powered by Discuz! X3.4