我们需要控制changeLog主题的大小，以降低Kafka存储成本。因此，我们在拓扑中使用transformer (DSL API)来调度一个标点符号，该标点符号使用keyValueStore.delete()从stateStore中删除旧记录。我能够验证在delete之后，在标点符号的进一步预定触发器上，被删除的键不在状态存储中</em
浏览 3提问于2020-08-29得票数 2
我已经使用streamsBuilder.table("myTopic")为一个主题创建了一个Ktable，并将其具体化到一个状态存储中，以便可以使用交互式查询。每隔一个小时，我希望从这个状态存储(以及相关的changelog主题)中删除其值在过去一个小时内没有更新的记录。我相信使用punctuator可以做到这一点，但到目前为止，我只使用了DSL，所以不确定<e
浏览 21提问于2020-07-20得票数 1
回答已采纳
1回答
我有一些用Java实现的Kafka用户，我正在实现一个独立的应用程序来检查记录和墓碑。希望卡夫卡将删除州商店，因为它压缩主题。卡夫卡删除相应主题中的旧记录时，是否删除了？当您在相应的主题中删除墓碑记录时，删除了吗？，我们是否坚持了？我看到的商店类型如下：
KSTREAM-AGGREGATE-S
浏览 4提问于2020-02-26得票数 2
1回答
我试图创建一个卡夫卡拓扑，并将其分解成更易读的。在按键分组时，我确实使用了.groupByKey(Grouped.with(Serdes.ByteArray(), Serdes.ByteArray()))。无论如何，我不知道如何解决这个问题。mergedTable.toStream() ...如何在不触发rocksdb存储异常的情况下将其分
浏览 7提问于2020-05-19得票数 0
回答已采纳
1回答
延缓卡夫卡流消费
、、、
我试图使用卡夫卡流(即不是简单的卡夫卡消费者)从重试主题中读取以前未能处理的事件。我希望使用retry主题，如果处理仍然失败(例如，如果外部系统关闭)，我希望将事件重新放到retry主题上。因此，我不想立即继续消费，而是在消费之前等待一段时间，以避免系统中充斥着暂时无法处理的消息。
return streamsBuilder.bui
浏览 2提问于2019-12-09得票数 6
我目前正在使用Kafka流作为有状态应用程序。状态是，而不是，存储在卡夫卡的状态存储中，而只是暂时存储在内存中。这意味着每当我重新启动应用程序时，所有状态都会丢失，并且必须通过从一开始就处理所有记录来重新构建它。在对Kafka状态存储进行了一些研究之后，这似乎正是我正在寻找的在应用程序重新启动(无论是在内存中还是在磁盘上)之间持
浏览 1提问于2020-01-13得票数 3
回答已采纳
2回答
我使用卡桑德拉数据库作为最终存储，而mongodb作为这个数据库的application.The工作流程的临时存储是，数据将在mongodb中存储3天，数据在mongodb存储一天后将存储在Cassandra注意，在从mongodb删除它的消息(3天后)之后，状态会出现，并理解没有迹象表明它是新消息或旧消息(已删除的消息)的状态。数据复制到卡桑德拉数据库后一天-(使用</e
浏览 13提问于2022-11-16得票数 -1
我有一个kafka streams拓扑，它从输入主题读取，更新一些状态，并确定状态条目是否需要保留在状态存储中，或者可以删除。如果它可以被删除，它将被删除，否则我有一个标点符号，它每10秒运行一次，并使状态存储中的项过期。
我最近发现，标点符号运行在相同的流线程上，可能会阻止流的处理。我可以使用哪些模式在单独的线程池中执行标点符号</em
浏览 10提问于2020-02-08得票数 2
在Kafka (或其他消息队列)中有没有一个功能，我们可以通过它来添加消息，这些消息只会在指定的时间变得活跃。这意味着该消息不应该在指定时间之前被消费。
浏览 0提问于2018-05-19得票数 0
2回答
当在kafka流中定义拓扑时，可以添加一个全局状态存储。它将需要一个源主题以及一个ProcessorSupplier。处理器接收记录，并在将它们添加到存储区之前从理论上对它们进行转换。但是在恢复的情况下，记录直接从源主题(changelog)插入全局状态存储，跳过最终在处理器中完成的转换。根据文件
注意:您不应该使用处理器将转换后的记录插入到全局状态</
浏览 2提问于2019-11-25得票数 18
回答已采纳
1回答
我使用kafkaStreams执行聚合，它实际上将所有聚合记录保存在一个keyValue状态存储中，该存储针对我正在生成的用于唯一标识该聚合的特定密钥。我不会使用任何卡夫卡窗口来进行这种聚合。如果我使用get，卡夫卡将在内部扫描完整的状态，并给我特定键的数据，所以如果我有一个键列表，它将遍历完整的statestore以获取列表中的键数。如果我设法为我的搜索键创建一个regex，我可
浏览 10提问于2022-06-23得票数 1
3回答
我在“掌握卡夫卡流和ksqlDB”一书中遇到了以下两个短语，作者使用了两个术语，它们的真正含义是“压缩主题”和“未压缩主题”。
表可以看作是对数据库的更新。在日志的此视图中，只保留每个键的当前状态(给定键的最新记录或某种聚合)。表通常是由压缩的主题构建的。在数据库术语中，流可以被看作是插入。每个不同的记录都保留在日志的这个视图中。
浏览 3提问于2020-07-21得票数 0
我希望将来自KStream的有窗口的输出分组起来，并将它们写到二级存储中。MainKStreamBuilder kStreamBuilder =
浏览 7提问于2016-08-31得票数 1
2回答
问题：我有一个卡夫卡主题的记录，需要从MySQL数据库中添加元数据(以及其他内容)。使用Flink，可以实现丰富的映射器，因此可以对多个记录重用连接。
transformers/处理器:用于有状态操作(很好！)，但也意味着使用状态存储(在我的
浏览 6提问于2017-04-04得票数 2
3回答
我有一个卡夫卡主题，我期望有两种不同的关键类型的信息:旧的和新的。即"1-new"、"1-old"、"2-new"、"2-old"。钥匙是独一无二的，但有些可能丢失了。现在，使用Kotlin和KafkaStreams API，我可以用来自新的和旧的相同的密钥id记录这些消息。joined.foreach({ key, value ->
log.info { "JOINED $key : $val
浏览 8提问于2018-01-10得票数 4
回答已采纳
我知道卡夫卡在内部使用RocksDb。如果我使用KTable存储数据字典，它会在特定的时间后被删除吗？如果是这样的话，我怎样才能让它持续无限的时间？
浏览 1提问于2018-05-10得票数 3
回答已采纳
1回答
我正在学习使用Apache的事件驱动架构和流系统。我了解了事件源和CQRS，并有一些关于实现的基本问题。我知道卡夫卡有一个持久化的状态存储可以维护所需的状态，但是我能像卡桑德拉表一样依赖它吗？有尺寸consideration?吗？中的驱动程序<
浏览 7提问于2022-09-12得票数 0
回答已采纳
我使用带有模式timestamp+incrementing的JDBC源连接器从Postgres获取表，使用Kafka。数据更新反映在卡夫卡主题中，但删除记录没有任何影响。所以，我的问题是：
有处理已删除记录的方法吗？如何处理已删除但仍然存在于卡夫卡主题中的记录？
浏览 5提问于2020-09-08得票数 2
回答已采纳
我正在使用一个GlobalKTable作为键值存储。存储中的数据很少变化，但确实会发生变化。我担心的是，该文件没有指定自上一个文件之后某个键何时被移除。在这种情况下，我不希望旧的、不再有效的密钥永远在应用程序存储中徘徊。c在所有应用程序的GlobalKTable存储中会发生什么？一旦主题从</
浏览 3提问于2020-11-26得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
