Количество фиксаций и смещений в каждом разделе темы кафки

Ответ 1

Из вашего вопроса не ясно, какой тип смещения вас интересует. На самом деле существует три типа смещений:

1. Смещение первого доступного сообщения в тематическом разделе. Используйте -2 (самый ранний) в качестве параметра --time для инструмента GetOffsetShell
2. Смещение последнего доступного сообщения в тематическом разделе. Используйте -1 (последний) в качестве параметра --time.
3. Смещение последнего прочитанного/обработанного сообщения, поддерживаемое потребителем kafka. Потребитель высокого уровня хранит эту информацию для каждой группы потребителей во внутренней теме Kafka (раньше называвшейся Zookeeper) и заботится о ее актуальности, когда вы вызываете commit() или когда для параметра автоматической фиксации установлено значение true. Для простого потребителя ваш код должен заботиться об управлении смещениями.

В дополнение к утилите командной строки, информация о смещении для # 1 и # 2 также доступна через SimpleConsumer.earliestOrLatestOffset().

Если количество сообщений не слишком велико, вы можете указать большой параметр --offsets для GetOffsetShell, а затем подсчитать количество строк, возвращаемых инструментом. В противном случае вы можете написать простой цикл в scala/java, который будет повторять все доступные смещения, начиная с самого раннего.

Из документации Кафка:

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.

Ответ 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, чтобы увидеть первое доступное сообщение.

Ответ 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 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.

Чтобы получить смещения для Topic_X для userGroup_Y, используйте команду ниже

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

Ответ 4

Эта информация также помогла создать script для просмотра количества сообщений в разделе для темы (из командной строки). Хотя такие инструменты, как Kafka-Web-Console, хороши, некоторые из нас живут в мире без GUI.

Вот script... используйте и модифицируйте его на свой страх и риск:-)

#!/bin/bash

topic=$1

if [[ -z "${topic}" ]] ; then

    echo "Usage: ${0} <topic>"
    exit 1

fi


if [[ -z "${KAFKA_HOME}" ]] ; then

    # $KAFKA_HOME not set, using default /kafka
    KAFKA_HOME="/kafka"

fi

if [ ! -d ${KAFKA_HOME} ] ; then

    echo "\$KAFKA_HOME does not point to a valid directory [$KAFKA_HOME]"
    exit 1

fi

cd $KAFKA_HOME

echo
echo "Topic: ${topic}: "

#
printf "Partition  Count\n"
printf "~~~~~~~~~~ ~~~~~~~~~~~~\n"

idx=0
for msg in `bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic ${topic} --broker-list localhost:9092 --time -1` ; do

    name=`echo ${msg} | awk -F ":" '{print $1}'`
    partition=`echo ${msg} | awk -F ":" '{print $2}'`
    total=`echo ${msg} | awk -F ":" '{print $3}'`

    printf "%10d %12d\n" ${partition} ${total}
    idx=$((idx + 1))

done

if [ ${idx} -eq 0 ] ; then

    echo "Topic name not found!"
    exit 1

fi

echo
exit ${rc}