Kafka设计（8）日志压缩

谁践踏了优雅 2023-07-21 05:53 11阅读 0赞

> 参考文档： http://kafka.apache.org/documentation/\#design

# 日志压缩 #

日志压缩确保Kafka总是保留一个主题分区的日志数据中每个消息key的至少最后一个已知值。它解决了一些用例和场景，例如：应用崩溃或系统错误后还原状态，在维护操作时重启应用后重新加载缓存。让我们更详细深入这些用例并描述压缩如何工作。

到目前为止，我们只描述了数据持久的简单方法（老数据在指定的时间周期前或当日志达到指定大小时被丢弃）。这对于临时事件日志（例如：每个记录都是独立的日志）可以很好的工作。但数据流中重要的一类中键值数据是可变的日志（例如：数据表的变更）。

让我们讨论流的具体例子。假如我们有一个包含用户email地址的主题，每次一个用户更新他的email地址时我们都发送一个消息到主题并使用用户id作为主键。现在假设我们在多个时间段为id为123的用户发送以下消息，每个消息都对应修改了email地址（其他用户的消息被忽略）：

123 => bill@microsoft.com
            .
            .
            .
    123 => bill@gatesfoundation.org
            .
            .
            .
    123 => bill@gmail.com

日志压缩带来了更颗粒状的保留机制，因此保证为每个主键保留最少保留最后一次更新。通过这些我们保证日志包含一个完整所有key值而不仅是最近更新key值的快照日志。这表示下游消费者可以根据此主题恢复他们自己的状态，无需恢复完整的包含所有变更的日志。

让我们先看一些有用的用例，并看看如何被使用：

1.  数据库变更订阅：通常有必要将数据集放到多个数据系统中，通常其中一个数据系统是某种类型的数据库（RDBMS或新流行的key-value存储）。例如，你有一个数据库、一个缓存、一个查询集群和一个Hadoop集群。每个数据库的改变都需要反映在缓存、查询集群以及最终到Hadoop。此时你只需要最近的日志就可以处理实时更新。但如果你想重新加载缓存或恢复宕机的查询节点，你仍旧需要完整的数据集。
2.  事件溯源（Event sourcing）：这是一类应用设计样式，它和查询处理(PS:CQRS)一同使用,使用变更的日志作为应用的主要存储。
3.  高可用日志：做本地计算的进程可以通过日志记录本地状态的变更以便其他进程可以重载变更并在在异常时接管来实现容错。一个具体的例子是在流查询系统中处理count、聚集或其他类似“group by”。Samza（一个实时流处理框架）正是出于这个目的使用[此功能][Link 1]。

在这些用例中每一个都主要需要处理实时的变更流，但偶尔当机器宕机或者数据需要重新加载或重新处理时需要做完全数据加载。日志压缩允许这些用例流从同一个主题中获取数据。这类日志应用在这个[博客站点][Link 2]有更详细描述。

通常的想法很简单，如果我们有无限的日志保留、记录上述案例的所有变更，那么我们可以捕获从系统一开始以来所有时间点的状态。使用完整日志，我们可以通过重放日志中前N条记录恢复到任意时间点。这个假想的完整日志对于会更新一条记录多次的系统来说很不实用，因为即使数据量很稳定，但日志将无边界的增长。如果采用丢失旧更新的日志保留机制，日志空间兼备限制，但将无法通过日志恢复到指定状态，因为旧更新不在日志中，无从从日志开头重新恢复到指定状态。

日志压缩是一种细粒度每个记录保留机制，而不是基于时间的粗粒度保留。我们选择了当相同主键有更新的更新时删除老记录的想法，这种方式的日志保证每个key都至少保留最后状态。

这个保留策略可以为每个主题设置，一个集群可以多个主题根据大小或时间保留而其他主题通过压缩保留。

这个功能受到Linkin的最早最成功的基础架构–名为[Databus][]的数据库变更日志缓存服务的启发。不像大多数日志结构的存储系统，Kafka为订阅和组织数据以便快速线性读写构建。不像Databus，Kafka可以作为事实源存储，因此可以在上游数据源无法重放时使用kafka。

### 日志压缩基本 ###

下面是一个高层视图展示了Kafka日志的逻辑结构，每个消息都有其偏移量。

![image][]

日志头和传统的Kafka日志一致。它有稠密的、顺序的偏移量并保留所有消息。日志压缩添加了用于处理日志尾部的选项。上图的日志中有被压缩的尾部。注意尾部的消息保留当它们初次写入时的原始偏移量（永远不变）。同样要注意，即使消息已被压缩，所有偏移量的位置依旧是有效的。在这个例子中，日志中偏移量的位置和下个更大偏移量的位置无法区分。例如，在上图中36、37、38偏移量都对应同样的位置，读取这三个偏移量任意一个都将返回38开始的消息集（PS:包含36、37、38三个消息）。

压缩同样允许删除。一个有key但无负载（payload）的消息被当作从日志中删除。这个删除标记可能导致所有带有这个key先前的消息被移除（带有这个key的新消息也一样），但具有删除标记的消息将在一段时间后从日志中清除以便释放空间。如上图所示，不再保留及时删除的点标记为“删除保留点”(PS:此点之前的数据都被删除无法读取，之后的数据可以被读取)。

压缩通过后台周期性重新拷贝日志段来完成。清理不会阻塞读取，并配置其使用不超过指定的IO吞吐量，以避免影响生产者和消费者。日志段压缩实际处理过程类似下图：  
![image][image 1]

### 日志压缩提供了什么保证 ###

日志压缩保证以下内容：

1.  任何停留在日志头部的消费者能够看到每个已写入的消息；这些消息拥有顺序的偏移量。
    
     *  主题的min.compaction.lag.ms参数定义了消息从写入到被压缩的最小的时间间隔。也就是说，它提供了消息保留在未压缩的头部的时间下限。
     *  主题的max.compaction.lag.ms参数用于保证消息写入到可以被压缩的最大延迟。
2.  消息的顺序永远被保留。压缩绝不会重排消息顺序只会移除一些消息。
3.  消息的偏移量不会改变。它在日志中是永久的位置标识。
4.  任何从日志起始开始处理的消费者将至少按顺序看到所有记录的最后状态。另外，假如消费者在主题delete.retention.ms设置的时间（默认24小时）内到达日志头部，所有已删除记录的删除标记也可以被看见。换句话说：由于移除删除标记和读取可以并发执行，有可能一个消费者在延迟超过delete.retention.ms时丢失删除标记。

### 日志压缩细节 ###

日志压缩被日志清理器（log cleaner）处理，这是一个后台线程池它重新拷贝日志段文件、移除在日志头部出现key的记录（PS：指日志头部出现相同key的新纪录，则尾部此key的老记录可以被删除）。每个压缩线程按以下所示工作：

1.  它选择日志头部和日志尾部比例最高的日志。
2.  它创建一个日志头部每个key最新偏移量的简明摘要。
3.  它重新从头到尾拷贝日志并删除之后还在日志中出现的key。新的已被清理的段被立即交换到日志以便只需要一个日志段的额外空间需求（而不是整个日志拷贝的空间）。
4.  日志头摘要本质上是一个空间压缩的hash表。它的每个条目精确的使用24个字节。一个清理迭代中只要8GB的清理缓存就可以清理大概366GB的日志头（假设有1千消息）

### 配置日志清理器 ###

日志清理器默认启用，将启动一个清理线程池。要在指定的主题启用日志清理可以通过以下添加日志特性属性设置

log.cleanup.policy=compact

log.cleanup.policy属性是broker配置定义在broker的server.properties文件。它影响集群中所有为设置覆盖该属性（[见文档][Link 3]）的主题。日志清理器可以配置为保留最小数量的未压缩头。通过设置压缩时间延迟启用

log.cleaner.min.compaction.lag.ms

这可以用于阻止主题中低于最小消息压缩时间的消息被压缩。如果没有设置，除最后日志段（即当前正在写入的日志段）外的所有日志段都可以被压缩。活动日志段不会被压缩即使所有消息都超过了最小压缩延迟时间。日志清理器可以配置以保证未压缩头可被压缩的最大延迟

log.cleaner.max.compaction.lag.ms

这可以用于防止很低产生效率的日志在无限制的周期内不被压缩。如果未设置，不会对超过min.cleanable.dirty.ratio比例的日志进行压缩。注意，压缩的最大期限不是强硬的保证，因为它是否被压缩还依赖于日志清理器线程是否可用以及实际的压缩耗时。你可以监控uncleanable-partitions-count、max-clean-time-secs和max-compaction-delay-secs指标。

更多的日志清理器配置在[此][Link 3]描述

[Link 1]: http://samza.apache.org/learn/documentation/0.7.0/container/state-management.html
[Link 2]: http://engineering.linkedin.com/distributed-systems/log-what-every-software-engineer-should-know-about-real-time-datas-unifying
[Databus]: https://github.com/linkedin/databus
[image]: /images/20230528/9aa5accdfd0d49c09751a214b3ac6e6d.png
[image 1]: /images/20230528/d554c05d36be40899dd657b43311e77f.png
[Link 3]: http://kafka.apache.org/documentation.html#brokerconfigs