kafka主题的每个分区中的提交和偏移量

Question

如何在已知 kafka主题的每个分区中查找提交数和当前偏移量.我正在使用kafka v0.8.1.1

Answer 1

你的问题不清楚,你感兴趣的是什么样的偏移.实际上有三种类型的偏移:

1. 主题分区中第一个可用消息的偏移量.使用-2(最早)作为GetOffsetShell工具的--time参数
2. 主题分区中最后一条可用消息的偏移量.使用-1(最新)作为--time参数.
3. 由kafka使用者维护的最后读取/处理的消息偏移量.高级消费者将此信息存储在Zookeeper中(分别针对每个消费者组),并在调用commit()或自动提交设置设置为true时注意使其保持最新.对于简单的消费者,您的代码必须注意管理偏移量.

除命令行实用程序外,#1和#2的偏移量信息也可通过SimpleConsumer.earliestOrLatestOffset()获得.

如果消息数量不是太大,您可以为GetOffsetShell指定一个大的--offsets参数,然后计算该工具返回的行数.否则,您可以在scala/java中编写一个简单的循环,它将从最早开始迭代所有可用的偏移量.

来自Kafka文档:

Get Offset Shell
get offsets for a topic
bin/kafka-run-class.sh kafka.tools.GetOffsetShell

required argument [broker-list], [topic]
Option Description 
------ ----------- 
--broker-list <hostname:port,..., REQUIRED: The list of hostname and hostname:port> port of the server to connect to. 
--max-wait-ms <Integer: ms> The max amount of time each fetch request waits. (default: 1000) 
--offsets <Integer: count> number of offsets returned (default: 1)
--partitions <partition ids> comma separated list of partition ids. If not specified, will find offsets for all partitions (default) 
--time <Long: timestamp in milliseconds / -1(latest) / -2 (earliest) timestamp; offsets will come before this timestamp, as in getOffsetsBefore  > 
--topic <topic> REQUIRED: The topic to get offsets from.

Answer 2

关于主题和分区的偏移量,您可以使用kafka.tools.GetOffsetShell.例如使用这些命令(我有主题games):

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic games --time -1

我会得到games:0:47841这意味着对于主题games和0分区我最近没有使用偏移47841(最新可用消息).

您可以使用-2查看第一个可用消息.

Answer 3

从 0.9.0.x 版本开始，您应该开始使用kafka.admin.ConsumerGroupCommand工具。以下是该工具采用的参数

List all consumer groups, describe a consumer group, or delete consumer group info.
Option                                  Description
------                                  -----------
--bootstrap-server <server to connect   REQUIRED (only when using new-
  to>                                     consumer): The server to connect to.
--command-config <command config        Property file containing configs to be
  property file>                          passed to Admin Client and Consumer.
--delete                                Pass in groups to delete topic
                                          partition offsets and ownership
                                          information over the entire consumer
                                          group. For instance --group g1 --
                                          group g2
                                        Pass in groups with a single topic to
                                          just delete the given topic's
                                          partition offsets and ownership
                                          information for the given consumer
                                          groups. For instance --group g1 --
                                          group g2 --topic t1
                                        Pass in just a topic to delete the
                                          given topic's partition offsets and
                                          ownership information for every
                                          consumer group. For instance --topic
                                          t1
                                        WARNING: Group deletion only works for
                                          old ZK-based consumer groups, and
                                          one has to use it carefully to only
                                          delete groups that are not active.
--describe                              Describe consumer group and list
                                          offset lag related to given group.
--group <consumer group>                The consumer group we wish to act on.
--list                                  List all consumer groups.
--new-consumer                          Use new consumer.
--topic <topic>                         The topic whose consumer group
                                          information should be deleted.
--zookeeper <urls>                      REQUIRED (unless new-consumer is
                                          used): The connection string for the
                                          zookeeper connection in the form
                                          host:port. Multiple URLS can be
                                          given to allow fail-over.

要为 consumerGroup_Y 获取 Topic_X 的偏移量，请使用以下命令

bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand --zookeeper <zookeeper urls> --describe --group consumerGroup_Y

响应看起来像

GROUP, TOPIC, PARTITION, CURRENT OFFSET, LOG END OFFSET, LAG, OWNER
consumerGroup, Topic_X, 0, 3030460, 3168412, 137952, none
consumerGroup, Topic_X, 1, 3030903, 3168884, 137981, none
consumerGroup, Topic_X, 2, 801564, 939540, 137976, none
consumerGroup, Topic_X, 3, 737290, 875262, 137972, none
consumerGroup, Topic_X, 4, 737288, 875254, 137966, none
consumerGroup, Topic_X, 5, 737276, 875241, 137965, none
consumerGroup, Topic_X, 6, 737290, 875251, 137961, none
consumerGroup, Topic_X, 7, 737290, 875248, 137958, none
consumerGroup, Topic_X, 8, 737288, 875246, 137958, none
consumerGroup, Topic_X, 9, 737293, 875251, 137958, none
consumerGroup, Topic_X, 10, 737289, 875244, 137955, none
consumerGroup, Topic_X, 11, 737273, 875226, 137953, none