获取Partition、Offset等主题详情的Kafka接口

文章/答案/技术大牛

发布

1回答

java、node.js、apache-kafka、kafka-consumer-api

我们有使用Kafka客户端的Node和Java应用程序。当前的需求是获取给定主题的每个分区的当前偏移量。我们能够在Confluent控制中心查看详细信息，并且需要在我们的应用程序中显示该信息。

浏览 37提问于2019-01-31得票数 0

2回答

使用python库检索kafka中的消费者组偏移量

python、apache-kafka、kafka-python

我有python脚本，并且需要使用kafka1 broker集群检索一组从kafka主题读取的消费者的当前消费者组偏移量。这些是原生kafka消费者，将偏移量存储在kafka集群中，而不是zookeeper中。脚本本身不需要使用任何消息，只需读取其他使用者的当前偏移量。我意识到使用kafka-consumer-groups.sh可以做到这一点，但理想情况下，我希望避免依赖外壳命令。我已经可以利用dpkp/kafka-python客户端做到这

浏览 1提问于2018-03-12得票数 3

1回答

不能在卡夫卡中创造超过15个主题

apache-kafka

我和我的同事在一个3节点集群上测试Kafka，我们遇到了这个问题，他们试图测试向多个主题发送消息的性能。我们不能创造超过15个主题。前15个主题工作得很好。但是，当试图创建第16个主题(和主题继续)，许多错误开始出现在两个追随者服务器。when processing the request (kafka.server.ReplicaFetcherThread)[2017-06-16 18:44

浏览 2提问于2017-06-16得票数 0

回答已采纳

1回答

Samza任务没有在一个分区上接收

apache-kafka、apache-samza

我有一个令人费解的问题，我的samza任务之一。除了一个分区上的消息外，它正常工作。我在这个主题上有9个分区。如果我发送1000条信息，我只收到其中的890条。我已经向kafka- console --使用分区键(我知道我的samza作业没有处理)检查过，控制台使用者确实看到了消息，所以我知道它被写到了主题上，并且至少有一个普通的用户可以看到它。使用kafkacat检查samza检查点主题，我看到分区4的</

浏览 2提问于2016-04-25得票数 2

回答已采纳

1回答

在一个集群中清除Kafka主题

apache-kafka

我知道我可以通过删除/data/ Kafka -log/ topic /*下的日志或将retention.ms配置设置为1000来清除代理上的kafka主题。我想知道如何清理多节点集群中的主题。

浏览 0提问于2018-06-26得票数 0

1回答

如何在partitionId应用程序中获取TopicName和KafkaStream

apache-kafka、apache-kafka-streams

如何从KafkaStream获取主题名称和分区id。对于任何其他的Kafka用户，我们可以获得主题名称和partitionId，如下所示： ConsumerRecords<String, String> records = consumer.pollConsumerRecord<String, String> record : records) {System.out.printf("consumed: key = %s, value = %s, part

浏览 2提问于2020-06-20得票数 4

回答已采纳

1回答

Kafka消费者无法读取Windows上的数据

java、apache-kafka

我正在尝试编写一个简单的Kafka消费者来读取windows机器上的数据。但是我的消费者不能读取任何数据。生产者产生了20多条消息，但没有一条消息被消费。ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); System.ou

浏览 0提问于2020-05-13得票数 1

1回答

不同的主题使用相同的偏移值。

apache-kafka、apache-storm

我们的拓扑使用KafkaSpout从kafka主题获取消息。我们有150个主题，包括12个分区，8个风暴执行者和2个风暴节点上的任务。风暴版本1.0.5，Kafka经纪人版本10.0.2，Kafka客户版本0.9.0.1。我们不会删除卡夫卡的主题。在某个时候，我在worker.log中观察到大量重复的警告消息。topic_1，partition=10}获取请求，

浏览 1提问于2018-05-31得票数 1

回答已采纳

1回答

Spark Structured Streaming:以批量查询的方式读取kafka主题中的前N条消息

apache-kafka、spark-structured-streaming

我有一个要求，我想要采样一个kafka主题(用于检查其数据质量等)，然后触发到它上的流作业。采样的参数之一可以是消息的数量。我引用了"http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#creating-a-kafka-source-for-batch-queries不可能从其中读取前N

浏览 32提问于2020-07-27得票数 1

1回答

不能将组中的kafka主题重置为1

apache-kafka

我有一个小组，有两个主题。我试图将其中一个偏移量重置为1，但得到的消息是偏移量(1)低于主题分区的最早偏移量。) is lower than earliest offset for topic partition mytopic-0.NEW-OFFSET当我尝试不同的分区时，我会得到相同的消息： ➜ local-kafka_2

浏览 0提问于2019-02-01得票数 1

1回答

为多个分区使用kafka批处理

apache-kafka、kafka-consumer-api、kafka-partition

我正在努力理解这种情况：，我有4个主题的，有一个消费者，卡夫卡将所有4个分区分配给它。，假设每批卡夫卡客户端从卡夫卡提取的5条消息。我在这里试图理解的是，如果一个批处理中的事件都来自同一个分区，然后循环到下一个分区批处理。还是批处理本身已经包含来自不同分区的事件？

浏览 4提问于2020-10-19得票数 3

回答已采纳

2回答

confluent_kafka:如何在读取数据之前可靠地查找(避免错误状态)

python、apache-kafka、confluent-kafka-python

我正在尝试将Python代码从confluent_kafka转换到，并且在读取历史数据时遇到了问题。该系统只有一个特定主题的生产者，以及几个独立的使用者(每个使用者都有一个单独的组ID)。当每个使用者启动时，它希望读取主题子集(称为历史主题)的最新历史消息，然后读取所有新消息。历史数据的确切起点并不重要，因为主要是获取很少编写的主题的信息。需要历史数据的

浏览 12提问于2022-08-23得票数 1

回答已采纳

1回答

(kafka版本为"0.8.2.2")

c++、apache-kafka

Kafka服务器版本= 0.8.2.2 std::string errstr;int64_t start_offset = RdKafka, start_<

浏览 1提问于2016-10-05得票数 1

回答已采纳

1回答

如何以编程方式获得Python中每个Kafka主题分区的最新偏移量

python、apache-kafka、kafka-topic

我是卡夫卡的新手，我想得到卡夫卡主题的位置，每个分区。我在文档( )中看到，偏移量可以通过函数KafkaAdminClient.list_consumer_group_offsets获得，但我看不到该位置的这种方法。有人知道我怎么弄到它吗？

浏览 9提问于2020-05-31得票数 3

回答已采纳

3回答

：如何对分区进行seekToEnd？

spring、spring-boot、apache-kafka、spring-kafka

, io.confluent.kafka.serializers.KafkaAvroDeserializer.class); 因此，我不需要获取字节，然后将它们反序列化为有效负载等等。key/value for partition ri00-q-log-et-final-0 at offset 36833.所以我决定从话题的结尾开始听，我检查了文档，它建议实现ConsumerSeekAware和它的子接口ConsumerSeekAware.ConsumerSe

浏览 0提问于2018-08-17得票数 0

回答已采纳

1回答

防止Kafka* JDBC接收器记录__connect_partition和__connect_offset*

jdbc、apache-kafka、apache-kafka-connect、confluent-platform

我在Kafka中有一个日志压缩的主题，它是通过JDBC接收器连接器写入Postgres的。尽管我在连接器上设置了mode=upsert，但它仍然在接收器数据库中为每个值添加了唯一行，因为它记录了每行的主题偏移量(__connect_offset)和分区(__connect_partition如何禁止JDBC接收器记录主题信息(我并不关心这些信息)？添加一个只获取我的数据列的fiel

浏览 0提问于2018-10-02得票数 2

1回答

使用Node.js + Kafka为消费者组创建分布式消费者

apache-kafka、kafka-consumer-api

我正在使用Node.js 10+，Apache 2.3和no Kafka npm软件包. console.log("Topic: " + topic, ", Partition: " + partition, ", Offset: " + m.offset</e

浏览 8提问于2019-12-23得票数 1

回答已采纳

2回答

如何使用kafka-node从topic中读取数据？

javascript、node.js、apache-kafka、producer-consumer

我有一个主题，我必须从kafka服务器读取，所以我只需要创建消费者，可以从kafka主题读取数据，我总是得到错误主题不存在。main.jsvar config = require('./con

浏览 3提问于2016-06-21得票数 2

回答已采纳

1回答

使用kafka-python消费kafka，seek()重置的本地偏移量会提交给kafka吗？

apache-kafka、kafka-python

使用kafka-python消费kafka，seek方法重置的本地偏移量会提交给kafka吗？我正在研究获取双中心机房Kafka集群的rpo索引的解决方案。使用kafka-python获取Kafka集群的最大时间戳，取两个机房的Kafka集群的最大时间戳之差。使用seek()将偏移量重置为分区的最大偏移量-1，

浏览 8提问于2021-05-20得票数 0

1回答

Kafka寻求从Python API开始？

python、apache-kafka、kafka-consumer-api、confluent-platform

有了Kafka Java API，我可以使用一个重新平衡侦听器，它会像这样定位到主题的开头(代码是Scala，Kafka API是Java)：} print

浏览 7提问于2017-03-03得票数 0

点击加载更多