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