1回答
Kafka:谁负责维护消费者群体读取偏移量消息的上限？
apache-kafka、kafka-consumer-api
我知道Kafka Queue Partition中的所有消息(或偏移量)都有它的偏移量，它负责偏移量的顺序。但是，如果我有一个Kafka Consumer Group (或单个Kafka Consumer)，它正在读取特别是Kafka主题分区，那么它如何维护读取的偏移量消息，以及谁维护这个偏移量计数器？如果消费
浏览 18提问于2020-07-04得票数 1
回答已采纳
2回答
有没有办法获取kafka主题的最早可用偏移量
java、apache-kafka、kafka-consumer-api、offset、spring-kafka
我有一个要求，在系统读取数据之前，我要存储偏移信息。因此，下次系统开始从kafka读取数据时，我需要将系统中的旧偏移量之间的数据读取到最新偏移量。但旧的偏移量可能由于kafka保留策略而无效。那么，如果我们在kafka消费者中指定较旧的偏移量，会有什么行为呢？另外，有没有办法获得特定主题/偏移
浏览 60提问于2020-09-09得票数 0
1回答
如何在Java中为消费群体拉取ConsumerOffset (存储在kafka中)？
java、apache-kafka、kafka-consumer-api
我有多个Kafka消费者和生产者，有不同的主题。使用一个独立的应用程序，我想要监控kafka消费者的滞后。我正在使用kafka 0.10.0.1，因为Kafka现在将消费者偏移量存储在kafka本身中，所以我如何才能读取相同的偏移量。
我能够读取每个分区的主题偏移量。
浏览 0提问于2017-03-06得票数 0
1回答
阅读卡夫卡在火花批次作业(fromOffset untilOffset)
scala、apache-spark、apache-kafka、kafka-consumer-api
我在这个上看到，我们可以使用org.apache.spark.streaming.kafka.KafkaUtils#createRDD在火花批处理作业中读取来自卡夫卡的消息，但是这种方法需要一个偏移范围，它需要一个“从偏移”和“直到偏移”。我从org.apache.spark.streaming.kafka.KafkaCluster#getLatestLeaderOffsets方法中获得了“from偏移量”，但是如何
浏览 0提问于2018-02-06得票数 0
1回答
Spark Streaming中驱动使用直接Kafka API查询分区偏移量的频率是多少？
apache-kafka、spark-streaming
是针对每个批处理间隔查询偏移量还是以不同的频率查询偏移量？
浏览 1提问于2015-11-17得票数 1
1回答
Kafka:同一群内多个消费者
apache-kafka、kafka-consumer-api
该集群中的每个应用程序都有一个客户端，该客户端订阅了一组相同的主题，这些主题在整个集群中都是相同的。此外，这些客户端中的每一个都共享相同的Kafka组ID。
现在，说到提交模式。我真的不想手动指定偏移量，但我也不想使用自动提交，因为在从Kafka收到数据后，我需要做一些处理。使用这种解决方案，我预计会出现“不同消费者接收的相同数据”问题，因为我在读取(消费)之前没有指定偏移
浏览 10提问于2018-02-01得票数 0
1回答
多连接器一主题分布式Kafka连接
elasticsearch、apache-kafka、apache-kafka-connect
分布式模式下Kafka Connect集群的Offset Management的行为是什么，即运行多个Connector并监听同一组主题(或一个主题)？因此，在分布式模式下，Kafka Connect将偏移量信息存储在Kafka中，该偏移量将被集群中的workers读取并提交。如果在该Kafka Connect集群中有多个Connector在监听同一主题，会发生什么情况？分区<em
浏览 15提问于2019-07-01得票数 2
1回答
使用KafkaUtils.createDirectStream在Kafka中存储消息偏移量
apache-spark、apache-kafka、spark-streaming-kafka
如果我使用KafkaUtils.createDirectStream读取消息，如何在Kafka中存储消息偏移量。每次应用程序运行down.It时，Kafka都会丢失偏移值，然后读取auto.offset.reset中提供的值(这是最新的)，并且在应用程序的停止-启动间隔内无法读取消息。
浏览 21提问于2019-01-14得票数 1
1回答
弹簧集成卡夫卡与管理补偿
java、spring、spring-integration、apache-kafka
我正在使用扩展来读取和处理Java应用程序中的Kafka消息。据我所知，它使用了高级的使用者API，它不允许完全管理动物园管理员中的偏移量。在我的例子中，我们有auto.commit.enable=false，以便在消息处理后将偏移量提交给。如果处理失败，那么将不会提交偏移量，我们应该尝试在配置好的时间内再次处理相同的消息，从的
浏览 1提问于2014-06-18得票数 3
1回答
仅从kafka偏移量获取一条记录
apache-kafka、kafka-consumer-api
我知道可以通过在kafka-console-consumer中使用--offset从kafka中读取特定偏移量的数据，但这会给出从该特定偏移量到当前偏移量的记录。如何才能将其限制为只有一条记录。
浏览 34提问于2021-06-16得票数 1
回答已采纳
1回答
使用Akka Stream和Kafka offset提交将事件从Kafka流式传输到Couchbase
scala、apache-kafka、akka、akka-stream、alpakka
我正在尝试设计一个Akka Stream，使用Alpakka来读取kafka主题中的事件，并将它们放到Couchbase中。，这个流实际上是从主题中读取事件，并将它们作为json文档放到Couchbase中，尽管我不知道如何将消费者偏移量提交给Kafka，但它看起来甚至很好。如果我已经清楚地理解了隐藏在Kafka消费者偏移量后面的主要思想，在任何故障或重启发生的情况下，流从最后提交的</em
浏览 21提问于2019-10-18得票数 0
2回答
卡夫卡的火花DStream总是从一开始就开始
apache-spark、apache-kafka、spark-streaming
查看我对解决方案的公认答案的最后评论 val kafkaParams = Map[String, Object](在Bhima Gogineni的回答中，我更改了我的配置如下：
val consumerParams
浏览 3提问于2018-07-04得票数 0
回答已采纳
1回答
手动重置kafka偏移量时，Flink偏移量进入不一致状态
apache-kafka、flink-streaming
我们有一个flink流应用程序，从卡夫卡读取消息。由于某些原因，我们不得不从kafka重置命令将kafka偏移重置为最新，因为有大量堆积。我们希望flink应用程序跳过所有这些消息，并从重置后出现的新消息开始。问题是因为flink在内部管理它的偏移量，它不知道这个重置，它现在只从后向读取msg (重置前的偏移点)，现在也不能提交偏移量。因此，每次重启flink应用程序时，它都会再次从同一点读
浏览 62提问于2021-06-19得票数 4
回答已采纳
2回答
如何在spark结构化流读取流中倒带Kafka偏移量
apache-spark、spark-structured-streaming
我有一个火花结构流作业，它被配置为从Kafka读取数据。请查看代码，带参数查看readStream()，读取Kafka的最新数据。我知道当一个新的查询开始而不是恢复时，readStream()会从第一个偏移量开始读取。
但我不知道如何在每次重新启动IntelliJ中的作业时启动新的查询。selectExpr("CAST(value as STRING)", "CAST(
浏览 0提问于2018-05-12得票数 3
1回答
Apache kafka偏移量在apache storm拓扑中提交
apache-kafka、apache-storm
我正在设计一个apache storm拓扑(使用streamparse)，它使用一个spout (apache kafka spout)和一个并行大于1的螺栓构建，可以批量读取kafka spout中的消息并将消息持久化到mysql表中当螺栓在mysql上插入失败时，我不会在kafka中提交偏移量
浏览 2提问于2019-01-31得票数 0
1回答
如何用火花结构流管理卡夫卡的偏移量
scala、spark-structured-streaming
我有一个火花结构化的流媒体工作，它需要读取kafka主题中的数据并进行一些聚合。如果我设置了startingOffsets="earliest"，那么作业将从主题读取所有数据，而不是从上一个流作业离开的位置读取数据。有人能帮助我如何配置，以设置偏移的权利，在最后一次流作业离开？我使用的是Spark2.4.0和kafka 2.1.1，我尝试为编写作业设置检查点位置，但似乎Spark
浏览 1提问于2019-05-15得票数 1
回答已采纳
1回答
我们如何手动重置通过spring boot java应用程序使用的kafka主题的偏移量？
java、spring-boot、apache-kafka、kafka-consumer-api
我的要求是，当应用程序无法处理从通过spring boot java应用程序使用的kafka主题的当前偏移量读取的消息时，重置kafka主题的偏移量。在手动重置偏移量或发送否定确认之后，需要通过spring boot java应用程序的kafka消费者再次从不合适的偏移量值轮询消息。它能实现吗？
浏览 3提问于2019-01-09得票数 1
2回答
使用python库检索kafka中的消费者组偏移量
python、apache-kafka、kafka-python
我有python脚本，并且需要使用kafka1 broker集群检索一组从kafka主题读取的消费者的当前消费者组偏移量。这些是原生kafka消费者，将偏移量存储在kafka集群中，而不是zookeeper中。
脚本本身不需要使用任何消息，只需读取其他使用者的当前偏移量。我意识到使用kafka-consumer-groups.sh可以做到这一点，但理想情况下，我希
浏览 1提问于2018-03-12得票数 3
1回答
informatica与Kafka的集成
apache-kafka、kafka-consumer-api、informatica
我计划在Informatica中使用Kafka数据对象。我想每5分钟通过Informatica从Kafka读取数据。我很少有疑问，卡夫卡如何认识到当前的偏移，从它必须加载。我们可以设置一个主题的偏移量，Informatica需要从它加载Kafka数据对象中的增量或更改的数据吗？我找到了以下答案，请您纠正我的确切过程: 1)动物园管理员是否将当前偏移量存储到加载数据的</em
浏览 2提问于2017-06-30得票数 0
5回答
Kafka主题和偏移量
apache-kafka
我有一个问题，假设有一个主题T1，有两个消费者C1，C2属于两个不同的组，当前偏移量是0。我们知道Kafka维护consumer.So的偏移量，如果C1消费消息，偏移量变成1，那么如果C2消费消息，它是从1偏移量开始还是从0偏移量开始消费消息？意味着如何维护两个不同消费者组的偏移量？
浏览 1提问于2020-02-20得票数 0
云服务器
对象存储
ICP备案
云点播
腾讯会议
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号