使用kafka-python消费kafka，seek方法重置的本地偏移量会提交给kafka吗？我正在研究获取双中心机房Kafka集群的rpo索引的解决方案。使用kafka-python获取Kafka集群的最大时间戳，取两个机房的Kafka集群的最大时间戳之差。使用seek()将偏移量重置为分区的最大偏移量-1，然后使用poll()获取最新的消息，但该消息无法在循环中获取，检查当前消
浏览 8提问于2021-05-20得票数 0
function getCurrentMessages(kafka, topic, cb_done){ var consumer= new kafka.Consumer(new kafka.Client(), [{topic: topic, offset: 0}], {fromOffset: true});
consumer.on
浏览 3提问于2015-10-03得票数 0
在给出任何细节之前，请注意，我不是，而是询问如何使用kafka-run-class.sh kafka.tools.ConsumerOffsetChecker从控制台获取最新的偏移量。kstream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRange) /*Doing Async Commit Here */我已经尝试了很多东西来获取给定主题的最新偏移量编辑:在上面的代码中，我试图通过以下方法获取<
浏览 1提问于2018-08-14得票数 0
回答已采纳
2回答
我正在开发一个使用kafka的scala应用程序。我的kafka消费者代码如下所示。") props.put() == offset) val value = data.value()} 在这里，我只想返回最新的值。当我运行我的应用程
浏览 53提问于2020-07-27得票数 1
我的kafka主题共有10条记录/消息和2个分区，每个分区有5条消息。我的使用者组有2个消费者，每个使用者已经分别从他们分配的分区中读取了5条消息。我停止了kafka使用者，并按照命令将使用者组偏移量重置为0。
./kafka-consumer-groups.sh --group cg1 --reset-offsets --to-offset 0 --topic t1 --execute --bootstrap-server在以后的测试周期中，我甚至没有运行上述命令来重置kafka使用
浏览 3提问于2019-06-27得票数 4
7回答
我正在为Kafka使用Python高级使用者，并想知道每个主题分区的最新偏移量。但是我不能让它起作用。from kafka import TopicPartition
con = KafkaConsumer(bootstrap_servers还是有一种不同的/更简单的方法来获得一个主题的最新偏移量？
浏览 6提问于2016-02-16得票数 33
回答已采纳
我的火花流应用程序从Kafka获取数据并对其进行处理。我找到了"auto.reset.offest"选项，它在星火中的行为并没有什么不同。然而，尽管它的行为出乎意料，它还是应该在删除后从最新的数据中获取数据。
但我发现不是。在使用数据之前，我看到所有的偏移都被清理干净了。然后
浏览 1提问于2014-08-26得票数 7
我使用Kafka 0.8.2从AdExchange接收数据，然后使用Spark Streaming 1.4.1将数据存储到MongoDB。它将继续阅读最新的offset of kafka，然后我将失去数据AdX推动卡夫卡在重新启动作业。我尝试了一些类似于auto.offset.reset -> smallest的东西，但是它最后会从0 ->接收到，那时数据是巨大的，在db中是重复的。如何将最新的offset火花保存到zookeeper或kafka</
浏览 5提问于2015-08-06得票数 14
回答已采纳
1回答
我使用的是Spotify docker版本的Kafka。我现在只有一个生产者、一个经纪人和一个消费者。这是我为消费者编写的代码，它不会打印来自Kafka的任何消息，即使那里有新消息。ascii')),
bootstrap_servers=['127
浏览 21提问于2018-08-16得票数 1
我正在尝试编写一个简单的Kafka消费者来读取windows机器上的数据。但是我的消费者不能读取任何数据。生产者产生了20多条消息，但没有一条消息被消费。ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
浏览 0提问于2020-05-13得票数 1
我想收到卡夫卡到Flink程序的最新数据，但是Flink正在阅读历史数据。我已经将auto.offset.reset设置为latest，如下所示，但是它没有工作Flink程序正在使用下面的代码从Kafka接收数据
//getting stream from Kafka and giving it assignTimestampsAndW
浏览 6提问于2017-08-23得票数 1
回答已采纳
在kafka手册网站中，有一个从kafka获取消息的代码块。但是，如果有一条大于获取大小的消息，则不起作用。import kafka.javaapi.consumer.SimpleConsumer;import <
浏览 4提问于2015-06-17得票数 0
回答已采纳
我对confluent_kafka非常陌生，但我已经获得了一些使用kafka-python的经验。我想要做的是改变开始消费消息的偏移量。说使用kafka-python包，我可以使用seek_to_end ()方法来获得最新提交的位置值。这样，我就可以使用seek方法()减去值并返回到前面的消息
另一方面，conflient_kafka似乎没有类似的函数，到目前为止，我所发现的是使用变量OFFSET_END，它的值为-1，它不会返回最新和最大的偏移量的数值我也可以使用&#
浏览 2提问于2018-04-23得票数 5
在使用消息时，我使用kafka-consumer-groups --describe --group test --bootstrap-server device2:9092来显示最新的CURRENT-OFFSET输出如下所示：comment 0 77042
浏览 2提问于2020-02-13得票数 1
回答已采纳
1回答
我在我的scala应用程序中使用kafka。我的代码如下。props.put("bootstrap.servers","localhost:9092") "org.apache.kafka.common.serialization.StringDeserializer") "org.apache
浏览 16提问于2020-07-22得票数 0
1回答
我希望只阅读使用kafka的火花流中最新的msg，但它也会获取过去的数据。如何在auto.offset.reset中设置KafkaUtil中的火花如何将conf设置为只获取当前消息。
浏览 3提问于2014-10-01得票数 4
回答已采纳
application.propertiesspring.kafka.consumer.auto-offset-reset=latest但预期的结果是，只对我发送的最新3个选项调用@KafkaListener方法。我试着使用另一种选择：但行为是一样的。
你能解释一下这些东西吗？它只依赖于spring.kafk
浏览 5提问于2017-10-16得票数 4
我试图从给定主题的每个分区获取最新的偏移量(未提交偏移量)。from kafka import KafkaConsumer, TopicPartition
broker = 'localhost:9092'
consumer.assign([tp]) #2last_offset= c
浏览 0提问于2019-04-24得票数 0
以下是我正在尝试的以及如何指示您已到达终点，或者不再有消息可供使用。如果偏移量与代理端偏移量不匹配，则如何做到这一点。
浏览 3提问于2020-04-21得票数 1
我需要订阅Kafka topic latest offset，阅读一些最新的记录，打印并完成。我如何在Spark中做到这一点？我想我可以做这样的事情 .read .option("kafka.bootstrap.servers", "192.168.1.1
浏览 4提问于2020-08-27得票数 0
云服务器
ICP备案
对象存储
腾讯会议
实时音视频
腾讯云开发者
