程序员经验分享-hwhale

用于重复数据删除的 Kafka 压缩

2023-12-01

我试图了解 Kafka 压缩的工作原理，并有以下问题：kafka 是否保证启用压缩的主题中存储的消息的键的唯一性？

Thanks!


简短的回答是否定的。

Kafka 不保证启用主题保留时存储的密钥的唯一性。

在 Kafka 中有两种类型cleanup.policy:

  • delete- 这意味着在配置的时间之后消息将不可用。有几个属性可以用于此目的：log.retention.hours, log.retention.minutes, log.retention.ms。默认情况下log.retention.hours is set 168。这意味着，消息older超过 7 天将deleted
  • compact- 对于每个键，至少有一条消息可用。在某些情况下可能是一个，但在大多数情况下会更多。压缩处理会定期在后台运行。它通过删除重复项并仅保留最后一个值来复制日志部分。

如果您只想读取每个键的一个值，则必须使用KTable<K,V>抽象自卡夫卡流.

关于键和压缩的最新值的相关问题：Kafka只订阅最新消息？

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

ApacheKafka

用于重复数据删除的 Kafka 压缩 的相关文章

  • 带有安全 Kafka 抛出的 Spark 结构化流：无权访问组异常

    为了在我的项目中使用结构化流 我正在 hortonworks 2 6 3 环境上测试 Spark 2 2 0 和 Kafka 0 10 1 与 Kerberos 的集成 我正在运行下面的示例代码来检查集成 我能够在 Spark 本地模式下的

  • Apache kafka - 消费者延迟选项

    我想在 Kafka 中为特定主题稍稍延迟启动一个消费者 具体来说 我希望消费者在从生成消息的时间起经过特定的时间延迟后开始使用该主题的消息 Kafka 中有任何属性或选项可以启用它吗 我们对火花流做了同样的事情 我希望 这种方法也适合您 这

  • 如何使用rest api设置kafka连接auto.offset.reset

    我创建了一个接收器 kafka 连接 将数据转换为其他存储 我想设置auto offset reset as latest当新连接器创建时kafka connect rest api 我已经设定consumer auto offset re

  • Apache Kafka 是否提供异步订阅回调 API？

    我的项目正在将 Apache Kafka 视为老化的基于 JMS 的消息传递方法的潜在替代品 为了让这个过渡尽可能的顺利 如果替代的排队系统 Kafka 有一个异步订阅机制那就更理想了 类似于我们当前项目使用的JMS机制MessageLis

  • kafka ProducerRecord 和 KeyedMessage 有什么区别

    我正在衡量卡夫卡生产者生产者的表现 目前我遇到了两个配置和用法略有不同的客户 Common def buildKafkaConfig hosts String port Int Properties val props new Proper

  • TopologyTestDriver 在 KTable 聚合上发送错误消息

    我有一个聚合在 KTable 上的拓扑 这是我创建的通用方法 用于根据我拥有的不同主题构建此拓扑 public static

  • Spark shell (spark 3.0.0) 添加包 confluence kafka 5.5.1 javax.ws.rs-api 问题

    我本地的win10 WSL回到ubuntu 在ubuntu上 我安装了spark3 0 0 confluence平台5 5 1 手动下载 当我尝试运行spark shell或spark submit时 下面是shell示例 spark sh

  • 将数据从 Kafka 存储传输到 Kafka 主题

    我想在卡夫卡做这样的事情 继续将数据存储在 KStream Ktable Kafka store 中 当我的应用程序收到特定事件 数据时 仅将上述存储中的特定数据集发送到主题 我们可以在卡夫卡中做到这一点吗 我认为单独使用 Kafka 消费

  • Kafka Streams - 如何扩展 Kafka 存储生成的变更日志主题

    我有多个冗余应用程序实例 它们想要使用主题的所有事件并独立存储它们以进行磁盘查找 通过rocksdb 为了便于论证 我们假设这些冗余消费者正在服务无状态 http 请求 因此 负载不是使用 kafka 共享的 而是使用 kafka 将数据从

  • 从 Apache Kafka 中的主题删除消息

    所以我是 Apache Kafka 的新手 我正在尝试创建一个简单的应用程序 以便我可以更好地理解 API 我知道这个问题在这里被问了很多 但是如何清除存储在主题上的消息 记录 我看到的大多数答案都说要更改消息保留时间或删除并重新创建主题

  • kafka消费者群体正在重新平衡

    我正在使用 Kafka 9 和新的 java 消费者 我正在循环内进行轮询 当代码尝试执行 Consumer commitSycn 时 由于组重新平衡 我收到 commitfailedexcption 请注意 我将 session time

  • Kafka：如何获取主题的最后修改时间，即添加到主题的任何分区的最后一条消息

    我们的用例是从 kafka 中删除陈旧 未使用的主题 即如果某个主题 在所有分区上 在过去 7 天内没有任何新消息 那么我们会将其视为陈旧 未使用并删除它 许多谷歌结果建议向消息添加时间戳 然后解析它 对于新主题和消息 灵魂可以工作 但我们

  • 如何检测 KTable 连接的哪一侧触发了更新？

    当您在 Kafka 中连接两个表时 每次更新两个 KTable 之一时 您的输出 Ktable 也会更新 想象一下你正在加入Customers与一个列表Orders你已经适当减少了 再次想象一下 您使用此连接的结果来为最终客户提供特别优惠和

  • 嵌入式 Kafka 测试随机失败

    我使用 EmbededKafka 实现了一系列集成测试 以测试使用 spring kafka 框架运行的一个 Kafka 流应用程序 流应用程序正在从 Kafka 主题读取消息 将其存储到内部状态存储中 进行一些转换并将其发送到另一个微服务

  • 如何使用 Python 在 Kafka 中生成 Tombstone Avro 记录？

    我的水槽属性 name jdbc oracle config connector class io confluent connect jdbc JdbcSinkConnector tasks max 1 topics orders con

  • 将 Kafka 输入流动态连接到多个输出流

    Kafka Streams 中是否内置了允许将单个输入流动态连接到多个输出流的功能 KStream branch允许基于真 假谓词进行分支 但这并不是我想要的 我希望每个传入的日志都确定它将在运行时流式传输到的主题 例如日志 date 20

  • Kafka 0.10 Java 客户端超时异常：包含 1 条记录的批次已过期

    我有一个单节点 多 3 个代理 Zookeeper Kafka 设置 我正在使用 Kafka 0 10 Java 客户端 我编写了以下简单的远程 在与 Kafka 不同的服务器上 生产者 在代码中我用 MYIP 替换了我的公共 IP 地址

  • 如何使用PySpark结构流+Kafka

    我尝试将 Spark 结构流与 kafka 一起使用 并且在使用 Spark 提交时遇到问题 消费者仍然从生产中接收数据 但 Spark 结构出错 请帮我找到我的代码的问题 这是我在 test py 中的代码 from kafka impo

  • 获取：导入 Spark 模块时出错：没有名为“pyspark.streaming.kafka”的模块

    我需要将从 pyspark 脚本创建的日志推送到 kafka 我正在做 POC 所以在 Windows 机器上使用 Kafka 二进制文件 我的版本是 kafka 2 4 0 spark 3 0 和 python 3 8 1 我正在使用 p

  • 是否有任何模拟器/工具可以生成流式传输消息？

    出于测试目的 我需要模拟客户端每秒生成 100 000 条消息并将它们发送到 kafka 主题 有没有任何工具或方法可以帮助我生成这些随机消息 有一个用于生成虚拟负载的内置工具 位于bin kafka producer perf test

随机推荐

  • Android 和 Java：减少服务循环上的内存使用

    我有一个 Android 服务 它使用此线程每秒更新一个通知 评论并不真正相关 thread new Thread Override public void run Preparando la notificaci n de Swap No

  • 字符串 s = 新字符串（“xyz”）。这行代码执行后创建了多少个对象？

    这个面试问题的普遍同意的答案是代码创建了两个对象 但我不这么认为 我写了一些代码来确认 public class StringTest public static void main String args String s1 a Stri

  • 如何在MySQL中记录记录的顺序集合

    假设我有一张桌子 里面有某种类型的记录 比如烹饪说明 比如 Fold the melted chocolate into the egg whites 该表包含唯一 ID 字段和字符串 我想为食谱构建另一个表 每个表都有一个唯一的 ID 和

  • 在对象内部使用“this”和对象名称进行引用有什么区别？

    如果我有以下代码 var obj x 34 init function alert this x alert obj x 两个警报都显示 34 但是有什么区别 一个比另一个更好吗 http jsfiddle net 4scz435q 我在j

  • 如何在 C++ 中编写正确的哈希表析构函数

    我正在写一个 C 哈希表 这是我的析构函数 HashMap HashMap for int i 0 i

  • 如何从 kdeplot 获取半高全宽 (FWHM)

    我在一些数据上使用了seaborn的kdeplot import seaborn as sns import numpy as np sns kdeplot np random rand 100 是否可以从创建的曲线返回 fwhm 如果不是

  • 教义 FindBy 方法与“OR 条件”？

    是否可以使用OR教义中的声明findBy method 我希望输出是这样的 SELECT FROM friends WHERE userId 1 OR FriendId 1 现在的代码 user repository gt findBy a

  • tf.where 的 TensorFlow 梯度在不应该返回 NaN 时返回 NaN

    下面是可重现的代码 如果运行它 您将看到在第一次 sess 运行中 结果为 nan 而第二种情况给出了正确的梯度值 0 5 但根据指定的 tf where 和条件 它们应该返回相同的值 我也根本不明白为什么 tf where 函数梯度在 1

  • 显示日志文件更新时的内容

    我有外部程序 例如 ffmpeg 和 gstreamer 在后台运行并写入日志文件 我想用我的 Flask 应用程序显示此日志的内容 以便用户可以观看日志更新 例如tail f job log会在终端做 我尝试使用指向日志文件 但未能显示数

  • pyspark中的DataFilter是什么？

    我看到一个叫做DataFilter在我的查询执行计划中 FileScan parquet product id 12 price 14 Batched true DataFilters isnotnull product id 12 For

  • 计时器：如何在后台保持计时器处于活动状态

    在我的 iPhone 定时器应用程序中 其中计时器应在后台运行 所以 我已经在 appdelegate 中设置了通知 它工作得很好 这样我就可以从视图控制器调用方法 这使得计时器处于活动状态 看一些代码 应用程序委托 void applic

  • h2混合模式连接问题

    我在 servlet 上下文侦听器中启动 h2 数据库 public void contextInitialized ServletContextEvent sce org h2 Driver load String apprealPath

  • 如何使用 proguard 获取发布构建 apk 文件

    我正在尝试使用ProGuard为了为我的项目制作发布 apk 文件 显然我正在使用许多第三方库 我只需要使用其中的几个类 我真的很想得到一些关于此的解释 我的调试版本超过20 MB 所以我想通过使用来减少它shrinking用于progua

  • NDB 查询 fetch() 和 ContextOptions

    我想仅在我的一个查询中禁用上下文缓存 我想我可以这样做 MyModel query ancestor user key fetch 100 options ContextOptions use cache False use memcach

  • HTML5 的 History.js - 需要进行黑客攻击才能不破坏 IE7

    我的目标是仅支持 HTML5 浏览器的 AJAX 历史记录 但是 我希望我的网站能够使用 HTML4 浏览器 但没有 AJAX 历史记录 许多 History js 示例在执行任何操作之前都包含以下检查 if History enabled

  • mailto链接多条正文线

    无法在 mailto 链接中使多行正常工作 就我而言 我正在使用 Outlook 默认邮件阅读器对其进行测试 以下内容放入锚点 href 中 mailto email protected subject test body type 20y

  • 如何防止 XmlSerialzer 转义“嵌套 XML”？

    我正在使用 XmlSerializer 来序列化 反序列化复杂对象 一个属性包含一个 XML 字符串 应将其写入字符串属性而不进行反序列化 示例 可在 LinqPad 中执行 XmlRoot RootObject Serializable

  • 从日期时间获取日期名称

    如何从 Python 中的日期时间对象获取日期名称 例如星期一 星期二 星期三 星期四 星期五 星期六和星期日 所以 举例来说 datetime 2019 9 6 11 33 0 应该给我 Friday import datetime no

  • 覆盖单个文件的编译标志

    我想使用一组全局标志来编译项目 这意味着我在顶级 CMakeLists txt 文件中指定了 ADD DEFINITIONS Wall Weffc pedantic std c 0x 但是 对于子目录中的特定文件 假设为 foo cpp 我

  • 用于重复数据删除的 Kafka 压缩

    我试图了解 Kafka 压缩的工作原理 并有以下问题 kafka 是否保证启用压缩的主题中存储的消息的键的唯一性 Thanks 简短的回答是否定的 Kafka 不保证启用主题保留时存储的密钥的唯一性 在 Kafka 中有两种类型cleanu

热门标签