我正在从事一个辅助项目，我将运输数据提取到 kafka 集群中。数据来自我所在城市的公共 API。例如：城市中的每条道路都通行。
我每隔几个小时就取一次道路工程。但是公共API没有返回timestamp，所以我无法轻易判断哪些道路工程是新的或最近被修改过的。大多数情况下，API 返回的内容自上次以来没有移动。
我使用 roadwork id 作为主题键并激活了日志压缩，所以有很多重复项并不会吓到我，因为我确信每项工作的最后状态都会被保留。
但是考虑到大量重复以及我只对最后一个版本感兴趣的事实，这样可以吗？我应该尝试检测新的/修改过的道路工程并只推动那些吗？有没有办法直接在 Kafka 中执行此操作？
【问题讨论】：
标签： apache-kafka kafka-producer-api
Kafka 的日志压缩非常适合您的用例。替代方案意味着您自己编写代码，同时增加额外的复杂性。
正如您已经注意到的，在启用日志压缩时，重要的是要记住至少每个键的最后状态（道路工作）都保存在主题中。您仍然会找到重复项。
为了尽量减少重复并因此保持较低的整体音量，您可以调整可用的topic configurations。最值得注意的是，我建议
降低 min.cleanable.dirty.ratio（默认为 0.5）以进行更频繁的清理。但是，请记住，这会导致清洁效率低下，同时使用更多资源。
减少max.compaction.lag.ms（默认为MAX_LONG）以减少消息在日志中不符合压缩条件的最长时间。
设置cleanup.policy=delete,compact，如果您的应用程序可以承受丢失较旧的消息。在该模式下，两个清理策略都将被激活，并且您可以在给定的保留时间（甚至字节大小）内至少保留每个键的最新状态。
此外，如果您关心卷大小，请在您的生产者中应用compression.type。从 Kafka 2.2.0 版开始，您可以使用 zstd，这通常有助于显着减少字节大小。
【讨论】：