как установить для потребителя в Faust заданное смещение c - PullRequest
Новая
       52

как установить для потребителя в Faust заданное смещение c

0 голосов
/ 17 февраля 2020

Из документации Faust я не могу узнать, как установить для потребителя заданное смещение c.

С confluent-kafka я использую consumer.offsets_for_times, чтобы найти start_offset, а затем назначить TopicPartition с указанным c смещением, что-то вроде: 

start_offset = consumer.offsets_for_times([
    TopicPartition("prediction.OfferPredictionCheckpoint", 0, int(start_date)),
    TopicPartition("prediction.OfferPredictionCheckpoint", 1, int(start_date)),
])

consumer.assign([
    TopicPartition("prediction.OfferPredictionCheckpoint", partition_number, pos)
])

С Faust я не могу найти намного больше, чем: 

consumer_auto_offset_reset

, которые позволяют устанавливать только самые ранние или самые последние. Как начать чтение с указанного c часа или начала дня?

1 Ответ

0 голосов
/ 19 февраля 2020

Я думаю, это может быть то, что вы ищете: https://faust.readthedocs.io/en/latest/reference/faust.transport.consumer.html#faust .transport.consumer.Consumer.seek

Может go с указанным c смещением Тем не менее, я не думаю, что есть простой способ сообщить ему go с указанным c временем или датой без каких-либо дополнительных лог c (может быть, бинарный поиск по-вашему, используя смещение?).

...