Kafka-压缩算法

Posted 2023-04-17

参考技术A 一、kafka压缩几点说明

首先说明一点kafka的压缩和kafka的compact是不同的，compact就是相同的key只保留一条，是数据清理方式的一种和kafka的定期删除策略是并列的；而kafka的压缩是指数据不删除只是采用压缩算法进行压缩。

kafka从0.7版本就开始支持压缩功能：

1）kafka的发送端将消息按照批量（如果批量设置一条或者很小，可能有相反的效果）的方式进行压缩。

2）服务器端直接将压缩消息保存（特别注意，如果kafka的版本不同，那么就存在broker需要先解压缩再压缩的问题，导致消耗资源过多）。

3）消费端自动解压缩，测试了下，发送端无论采用什么压缩模式，消费端无论设置什么解压模式，都可以自动完成解压缩功能。

4）压缩消息可以和非压缩消息混存，也就是说如果你kafka里面先保存的是非压缩消息，后面改成压缩，不用担心，kafka消费端自动支持。

二、kafka压缩算法种类和性能区别

测试的kafka版本：kafka_2.12-1.1.1

测试的kafka客户端版本：0.10.2.1

测试数据的条数：20000

kafka支持三种压缩算法，lz4、snappy、gzip，

压缩算法大小压缩比生成耗时（毫秒）消费耗时（毫秒）

None8.2MB0473917464

gzip1.9 MB23%468416257

snappy3.2 MB39%393616726

lz42.9 MB35%372317229

测试遇到疑问，开始非压缩算法发送2万条大小为16MB，后面再发送到gzip的时候大小竟然自动变成了8.2MB，采用的是delete模式，估计可能是日志之类的，snappy也有类似的现象开始是4.0MB，后面log大小缩小为3.2MB，有朋友知道麻烦告知。

我怀疑可能是版本原因导致数据重新被压缩，1.1.1优化的更好，所以压缩效果更好

通过上面数据来看，gzip的压缩效果最好，但是生成耗时更长，snappy和lz4的数据差不多，更倾向于lz4，具huxi大神的书上所说kafka里面对snappy做了硬编码，所以性能上最好的是lz4，推荐使用此压缩算法。

压缩率对比：

图: http://i.stack.imgur.com/LPCSe.png

性能对比图：

网上找来图

压缩设置

很简单：

/ compressType有四种取值:none lz4 gzip snappy /props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,compressType);

其他说明

消费端无论设置什么压缩模式，都可以正确的解压kafka的消息，也就是说消费端可以不设置解压缩，

不过可能性能有所下降。

如何测试日志压缩在 Kafka 中是不是有效？

【中文标题】如何测试日志压缩在 Kafka 中是不是有效？

【英文标题】：How to test whether log compaction is working or not in Kafka?如何测试日志压缩在 Kafka 中是否有效？

【发布时间】：2016-03-20 19:06:27

【问题描述】：

我已经在 Kafka 0.8.1.1 中的 server.properties 文件中进行了更改，即添加了 log.cleaner.enable=true 并在创建主题时启用了 cleanup.policy=compact。 现在，当我测试它时，我使用以下 (Key, Message) 将以下消息推送到主题。

偏移量：1 - (123, abc)； 偏移：2 - (234, def); 偏移：3 - (345, ghi); 偏移量：4 -（123，已更改）

现在我使用与先前输入相同的键推送第四条消息，但更改了消息。这里应该出现日志压缩。并且使用 Kafka 工具，我可以看到主题中的所有 4 个偏移量。我如何知道日志压缩是否有效？是否应该删除较早的消息，或者在推送新消息时日志压缩工作正常。 它与log.retention.hours 或topic.log.retention.hours 或log.retention.size 配置有什么关系吗？这些配置在日志压缩中的作用是什么。 附言- 我已经彻底阅读了 Apache 文档，但仍然不清楚。

【参考方案1】：

实际上，日志压缩只有在日志数量达到非常高的计数（例如 100 万）时才可见。所以，如果你有这么多的数据，那很好。否则，使用配置更改，您可以将此限制减少到 100 条消息，然后您可以看到在具有相同键的消息中，只有最新的消息会出现，之前的消息将被删除。如果您每次都有数据的完整快照，最好使用日志压缩，否则您可能会丢失具有相同关联键的先前日志，这可能很有用。

【讨论】：

如何通过Kafka提供的命令行工具来做到这一点？

【参考方案2】：

尽管这个问题是几个月前的问题，但我只是在为我自己的问题进行研究时遇到它。我创建了一个最小的示例来了解压缩如何与 Java 一起工作，也许它对你也有帮助：

https://gist.github.com/anonymous/f78184eaeec3ee82b15182aec24a432a

此外，查阅文档后，我在主题级别使用以下配置以尽快启动压缩：

min.cleanable.dirty.ratio=0.01
cleanup.policy=compact
segment.ms=100
delete.retention.ms=100

运行时，该类显示压缩有效 - 在主题上只有一条具有相同键的消息。

通过适当的设置，这将可以在命令行上重现。

【讨论】：

如果你最后没有填写 1 segment (Topic->Partition->Segment) 我会补充一点，不会发生压缩。假设 segment.ms=100 并且您在测试的前 100 毫秒内写入 segement.bytes=1GB，那么您将只有 1 个段，并且您的数据不会被压缩。请确保您在比segment.ms 更长的时间内写入同一分区，或者将segment.bytes 设置为较低的值，以便并非所有数据都适合1 个段。

强制每 100 毫秒滚动一次片段对于测试来说是可以的，但在生产中会非常低效。默认值为 604800000 毫秒（7 天）。

【参考方案3】：

最好也看看log.roll.hours，默认情况下是 168 小时。简而言之：即使您没有如此活跃的主题并且您无法在一周内填充最大段大小（默认为普通主题 1G，偏移主题默认为 100M），您将拥有一个大小低于@的封闭段987654322@。该段可以在下一轮压缩。

【参考方案4】：

要从 CLI 检查主题属性，您可以使用 Kafka-topics cmd 进行：

https://grokbase.com/t/kafka/users/14aev0snbd/command-line-tool-for-topic-metadata

【讨论】：

这可能会显示cleanup.policy=compact，但不能验证主题实际上得到压缩。

【参考方案5】：

您可以使用 kafka-topics CLI 来完成。 我从 docker(confluentinc/cp-enterprise-kafka:6.0.0) 运行它。

$ docker-compose exec kafka kafka-topics --zookeeper zookeeper:32181 --describe --topic count-colors-output
Topic: count-colors-output  PartitionCount: 1   ReplicationFactor: 1    Configs: cleanup.policy=compact,segment.ms=100,min.cleanable.dirty.ratio=0.01,delete.retention.ms=100
    Topic: count-colors-output  Partition: 0    Leader: 1   Replicas: 1 Isr: 1

但如果您在 Config 字段中没有看到任何内容，请不要感到困惑。如果使用默认值，就会发生这种情况。因此，除非您在输出中看到 cleanup.policy=compact - 主题不会被压缩。