Как получить последнее смещение для раздела для темы kafka?

Я использую Python высокого уровня для Kafka и хочу знать последние смещения для каждого раздела темы. Однако я не могу заставить его работать.

from kafka import TopicPartition
from kafka.consumer import KafkaConsumer

con = KafkaConsumer(bootstrap_servers = brokers)
ps = [TopicPartition(topic, p) for p in con.partitions_for_topic(topic)]

con.assign(ps)
for p in ps:
    print "For partition %s highwater is %s"%(p.partition,con.highwater(p))

print "Subscription = %s"%con.subscription()
print "con.seek_to_beginning() = %s"%con.seek_to_beginning()

Но вывод, который я получаю,

For partition 0 highwater is None
For partition 1 highwater is None
For partition 2 highwater is None
For partition 3 highwater is None
For partition 4 highwater is None
For partition 5 highwater is None
....
For partition 96 highwater is None
For partition 97 highwater is None
For partition 98 highwater is None
For partition 99 highwater is None
Subscription = None
con.seek_to_beginning() = None
con.seek_to_end() = None

У меня есть альтернативный подход с использованием assign, но результат тот же

con = KafkaConsumer(bootstrap_servers = brokers)
ps = [TopicPartition(topic, p) for p in con.partitions_for_topic(topic)]

con.assign(ps)
for p in ps:
    print "For partition %s highwater is %s"%(p.partition,con.highwater(p))

print "Subscription = %s"%con.subscription()
print "con.seek_to_beginning() = %s"%con.seek_to_beginning()
print "con.seek_to_end() = %s"%con.seek_to_end()

По некоторым данным, я могу получить такое поведение, если fetch не был выпущен. Но я не могу найти способ заставить это. Что я делаю неправильно?

Или есть другой/более простой способ получить последние смещения для темы?

Ответы

Ответ 1

Наконец, проведя день на этом и несколько ложных запусках, я смог найти решение и заставить его работать. Отправьте ее ей, чтобы другие могли ссылаться на нее.

from kafka import SimpleClient
from kafka.protocol.offset import OffsetRequest, OffsetResetStrategy
from kafka.common import OffsetRequestPayload

client = SimpleClient(brokers)

partitions = client.topic_partitions[topic]
offset_requests = [OffsetRequestPayload(topic, p, -1, 1) for p in partitions.keys()]

offsets_responses = client.send_offset_request(offset_requests)

for r in offsets_responses:
    print "partition = %s, offset = %s"%(r.partition, r.offsets[0])

Ответ 2

Если вы хотите использовать сценарии оболочки Kafka, присутствующие в kafka/bin, вы можете получить самые последние и самые низкие смещения, используя kafka-run-class.sh.

Для получения последней команды смещения будет выглядеть следующим образом:

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

Чтобы получить самую маленькую команду смещения, будет выглядеть следующим образом:

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

Дополнительную информацию о Get Offsets Shell можно найти в следующей статье ссылка

Надеюсь, это поможет!

Ответ 3

from kafka import KafkaConsumer, TopicPartition

TOPIC = 'MYTOPIC'
GROUP = 'MYGROUP'
BOOTSTRAP_SERVERS = ['kafka01:9092', 'kafka02:9092']

consumer = KafkaConsumer(
        bootstrap_servers=BOOTSTRAP_SERVERS,
        group_id=GROUP,
        enable_auto_commit=False
    )


for p in consumer.partitions_for_topic(TOPIC):
    tp = TopicPartition(TOPIC, p)
    consumer.assign([tp])
    committed = consumer.committed(tp)
    consumer.seek_to_end(tp)
    last_offset = consumer.position(tp)
    print("topic: %s partition: %s committed: %s last: %s lag: %s" % (TOPIC, p, committed, last_offset, (last_offset - committed)))

consumer.close(autocommit=False)

Ответ 4

Другим способом достижения этого является опрос потребителя для получения последнего потребленного смещения, а затем использование метода seek_to_end для получения последнего доступного смещенного раздела.

from kafka import KafkaConsumer
consumer = KafkaConsumer('my-topic',
                     group_id='my-group',
                     bootstrap_servers=['localhost:9092'])
consumer.poll()
consumer.seek_to_end()

Этот метод особенно полезен при использовании групп потребителей.

ИСТОЧНИКИ:

Ответ 5

С kafka-python>=1.3.4 вы можете использовать:

kafka.KafkaConsumer.end_offsets (разделы)

Получить последнее смещение для заданных разделов. Последнее смещение раздела - это смещение предстоящего сообщения, то есть смещение последнего доступного сообщения + 1.

from kafka import TopicPartition
from kafka.consumer import KafkaConsumer

con = KafkaConsumer(bootstrap_servers = brokers)
ps = [TopicPartition(topic, p) for p in con.partitions_for_topic(topic)]

con.end_offsets(ps)