腾讯云
开发者社区
文档
建议反馈
控制台
首页
学习
活动
专区
工具
TVP
最新优惠活动
文章/答案/技术大牛
搜索
搜索
关闭
发布
登录/注册
精选内容/技术社群/优惠产品,
尽在小程序
立即前往
文章
问答
(6722)
视频
沙龙
1
回答
Kafka
Streams
-
覆盖
默认
的
addSink
实现
/
自定义
生产者
这是我在这里发表
的
第一篇文章,我不确定以前是否在这里讨论过这个问题,但这里是这样
的
:我有一个使用处理器API
的
Kafka
Streams
应用程序,遵循以下拓扑结构: 1.Produce its process status to an output topic (processor.
addSink
()) 然而,为了可追溯性
的
目的,我需要在日志中记录kstreams向输出主题生成消息
的
时刻KEY="MY_KEY" OUTPUT_TOP
浏览 47
提问于2020-04-30
得票数 1
回答已采纳
1
回答
如何在属性文件弹簧引导中设置Acks=all
、
、
我想在我
的
春季云流卡夫卡应用程序中为我
的
生产者
设置acks=all属性。我试过这样做:和和 spring.cloud.stream.
kafka
.
streams
.bindings.<channel>.producer.con
浏览 1
提问于2019-04-24
得票数 1
2
回答
流处理器(低级API)源处理器如何从topic获取数据?
、
我是
kafka
流处理器
的
新手,偶然发现了“拓扑”
的
关键概念。如下所示: Topology topology = new Topology(); topology.addSource("SOURCE", "source-topic"); 上面的代码片段将创建(如果我
的
理解正确的话)一个名为" source“
的
源流处理器,并将侦听
kafka
主题"source- topic”。我没有为这个“源”流处理器写任何代码,它怎么能从
k
浏览 32
提问于2019-03-19
得票数 1
回答已采纳
1
回答
在
Kafka
Streams
中,即使无法发布到输出主题,消费者偏移量也会提交吗?
、
如果我有一个
Kafka
流应用程序无法发布到主题(因为主题不存在),它是提交消费者偏移量并继续,还是在相同
的
消息上循环,直到它可以解析输出主题?根据我
的
观察,这个应用程序只是打印了一个错误,并且运行正常。} 在我看来,为了不丢失数据,它只会在相同
的
消息上旋转,直到问题得到解决?关于
默认
行为是什么,我找不到明确
的
答案。我们没有将自动提交设置为off或类似的设置,大多数设置都设置为
默认
值。我这样问是因为我们不想在健康检查正常
的
情况下结束(应用程序正在运行,同
浏览 13
提问于2019-05-31
得票数 1
回答已采纳
1
回答
使用
kafka
流处理器api处理失败消息
、
使用
kafka
流处理器api topology.addProcessor("StreamsProcessor",()->new StreamsProcessor(), "mySource"); topology.
addSink
("sink1","outpu
浏览 2
提问于2020-06-26
得票数 0
2
回答
具有
默认
分区
的
Kafka
生产者
、
现在,我
的
kafka
生成器正在将所有消息下沉到
kafka
主题
的
单个分区中,该主题实际上具有超过1个分区。Properties props = new Properties();props.putProducerConfi
浏览 0
提问于2018-08-28
得票数 0
2
回答
Flink
kafka
- Flink作业不向不同分区发送消息
、
、
我
的
配置如下: speStream.
addSink
(new FlinkKafkaProducer011(kafkaTopicName,new 场景1: 我在eclipse上写了一个flink作业(
生产者
),它从文件夹中读取一个文件,并将msgs放在
kafka
主题上。例如,
浏览 0
提问于2018-08-29
得票数 2
回答已采纳
4
回答
处理
Kafka
流中
的
异常
、
、
已经经历了多个帖子,但大多数都是相关
的
,处理坏消息,而不是在处理它们时处理异常。 如何处理重试?
浏览 0
提问于2018-07-12
得票数 26
回答已采纳
1
回答
如何
实现
FlinkKafkaPartitioner?
、
、
我正在开发一个Flink服务,在这里我读取来自一个
Kafka
的
消息,然后将它们反序列化为HashMap,并对它们做一些事情,最后将它们写到另一个
Kafka
。我想要做
的
是为我
的
Flink
Kafka
生产者
创建一个
自定义
分区,这样具有相同id
的
事件就会转到同一个分区,因为保持事件与同一个id保持顺序是很重要
的
。到目前为止,我对
生产者
的
了解如下(对FlinkKafkaProducer使用
浏览 0
提问于2019-11-12
得票数 0
回答已采纳
2
回答
Kafka
Streams
- java版本依赖关系
我正在使用Java1.6运行我
的
应用程序,并计划将一些方面与
Kafka
集成。 我正在探索是否应该使用
Kafka
Consumer (使用poll())或使用KafkaStreams API。
浏览 3
提问于2018-12-20
得票数 0
2
回答
kafka
在单个事务中生成主题并写入状态存储
、
、
是否可以在单个事务中生成
Kafka
主题并写入状态存储?而不是作为主题消费
的
一部分启动事务。 编辑:我这么做
的
原因是为了能够过滤掉重复
的
请求。我也知道Confluent提供
的
解决方案。但是,只要“从主题到商店”
的
同步时间小于阻塞时间,这就可以工作。
浏览 3
提问于2019-03-01
得票数 0
2
回答
在
kafka
中处理数据流
我正在尝试设置一个
kafka
消费者来处理来自
Kafka
流
的
数据。我能够设置到流
的
连接,数据是可见
的
,但它是特殊字符和ASCII
的
混合。我正在使用内置
的
kafka
控制台,但也尝试了python版本
的
confluent-
kafka
。唯一需要遵循
的
参数是将SASL_PLAINTEXT安全协议与SCRAM-SHA-256一起使用。
Kafka
控制台 bin/
kafka</e
浏览 37
提问于2019-04-08
得票数 0
1
回答
卡夫卡流与JoinWindow合作进行数据重放
、
假设我想为我过去6个月
的
数据重新做这些连接,因为我正在为所有数据同时运行管道,kafkaStream将加入所有可连接
的
数据,而且它没有考虑时间差(它应该只加入超过一个月
的
数据)。我假设JoinWindow时间是我们将数据插入卡夫卡主题
的
时候,对吗?这一次我如何修改和操作,以便正确地运行我
的
数据回放,我
的
意思是,要重新插入过去6个月
的
数据,每个记录都需要一个月
的
窗口,并加入一个基于该记录
的
窗口。这个问题不是的重复,在那里我问我如何能够根
浏览 4
提问于2017-01-23
得票数 4
回答已采纳
1
回答
Quarkus -
Kafka
流-如何将配置选项传递给底层
生产者
和消费者
、
流应用程序
的
必要配置选项,例如,必须将quarkus.
kafka
-
streams
.bootstrap-servers=localhost:9092插入到包含项目的application.propertiesQuarkus还为更精细
的
配置提供了一个传递选项。这些文件指出:
kafka
-
streams
命名空间中
的
所有属性都通过同样
的
方式传递给
Kafka
Streams
引擎。更改它们
的
值需要重新构建应用
浏览 8
提问于2022-04-08
得票数 1
回答已采纳
3
回答
kafka
消费者可以在轮询来自一个主题
的
所有消息之前对消息进行过滤吗?
、
、
、
这意味着我们必须消费/接收来自主题
的
所有消息,并在客户端对其进行过滤。 太贵了。我想知道我们是否可以根据一些已经传递给代理
的
东西来过滤和接收特定类型
的
消息,比如msg密钥或其他东西。从Consumer.poll(timeout)方法来看,我们似乎不能做额外
的
事情。
浏览 1
提问于2018-06-27
得票数 11
1
回答
Kafka
Streams
RoundRobinPartitioner
、
、
我写了一个
kafka
streams
代码,使用
kafka
2.4
kafka
客户端版本和
kafka
2.2服务器版本。我有50个分区在我
的
主题和内部主题。我
的
kafka
流代码有selectKey() DSL操作,我有两百万条使用相同密钥
的
记录。如果我没有像预期
的
那样使用循环调度,我
的
所有消息都会转到同一个分区。 直到现在一切都还好,但我意识到了;当我使用RoundRobinPartitioner类时
浏览 49
提问于2020-01-08
得票数 1
1
回答
防止__TypeId__在Spring流中
的
使用
、
我们有一个流氓
生产者
将
Kafka
头__TypeId__设置为
生产者
的
一部分,但不是使用
Kafka
流绑定器在Stream应用程序中
实现
的
消费者。bootstrap-servers: 'xxxxx.
kafka
.dev.dvint.de:9093' stream:
streams
::569) at org.apache.
浏览 4
提问于2022-01-25
得票数 0
回答已采纳
1
回答
金刚卡夫卡溪流
、
我正在尝试使用golang在Go中创建一个
kafka
流客户端。据我所见,这只有在使用Java客户机时才有可能。我做了一些搜索,并找到了其他一些第三方图书馆,但没有正式
的
。另外,根据我有限
的
理解,我认为流是语法糖而不是标准消费者吗?这是对
的
吗?
浏览 4
提问于2021-12-13
得票数 1
1
回答
如何保证
Kafka
流聚合当天
的
数据
、
、
我有一个流应用程序,计算高/低/交易量
的
传入交易。我有一个有交易信息
的
输入主题。没有与这些消息相关联
的
时间戳。我聚合数据并计算高/低/音量,然后创建一个名为HiLowMessage
的
新消息并将其流出到我
的
输出流中。 因为输入主题总是有数据。我如何确保只对今天
的
数据进行聚合,而不包括昨天
的
数据?
浏览 14
提问于2019-11-14
得票数 1
1
回答
如何在序列化程序类中调用
默认
方法
、
、
我一直试图通过扩展类来编写
自定义
序列化程序。 return serialize(topic, data);但是,每当我为序列化器编写
自定义
实现
时,它都会调用被
覆盖
的
方法。byte[] serialize(String topic, T data); Iam将这个类传递给
kafka
生成器,用于序列化程序,因此
浏览 1
提问于2021-03-15
得票数 0
回答已采纳
点击加载更多
扫码
添加站长 进交流群
领取专属
10元无门槛券
手把手带您无忧上云
相关
资讯
Apache Kafka 3.0 发布,离彻底去掉ZooKeeper更进一步
Kafka 2.0正式发布,带来众多改进
系统解读Kafka的流和表(二):主题、分区和存储
kafka实战演练,欲走大数据之路的台阶打造,通往的必经之路
Apache Kafka 2.0.0版本发布,多方面功能得到加强
热门
标签
更多标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM
活动推荐
运营活动
广告
关闭
领券