文章/答案/技术大牛

发布

社区首页 >问答首页 >如何将Faust中的消费者设置为特定的偏移量

问如何将Faust中的消费者设置为特定的偏移量
EN

Stack Overflow用户

提问于 2020-02-18 01:07:48

回答 2查看 1.5K关注 0票数 3

从Faust文档中，我找不到如何将消费者设置为特定的偏移量。

使用confluent-kafka时，我使用consumer.offsets_for_times查找start_offset，然后将TopicPartition分配给该特定偏移量，如下所示：

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)
])

关于浮士德，我找不到更多：

consumer_auto_offset_reset

它只允许您设置最早或最晚。我该如何从特定的小时或一天的开始开始阅读呢？

faust

回答 2

Stack Overflow用户

回答已采纳

发布于 2020-02-20 02:25:52

我想这可能就是你要找的：https://faust.readthedocs.io/en/latest/reference/faust.transport.consumer.html#faust.transport.consumer.Consumer.seek

它可以转到特定的偏移量，但是我不认为有一种简单的方法可以告诉它转到特定的时间或日期，而不需要一些额外的逻辑(也许可以使用偏移量进行二进制搜索？)。

票数 1

Stack Overflow用户

发布于 2021-05-28 17:27:18

要将偏移量设置为特定值，可以使用以下示例。这里我将偏移量设置为50000。每次我启动我的应用程序时，代理程序从偏移量50000开始读取。为此，我使用app.consumer.seek

这里，tp接受两个参数，在本例中为topic - test，分区号为0。有关更多信息，请访问faust.types

from faust.types import TP, Message

tp = TP("test", 0)
topic = app.topic(tp.topic)

@app.task()
async def on_start():
    await app.consumer.seek(tp, 50000)
    print("App startet")

@app.agent(topic)
async def receive(stream):
    async for event in stream.events():
        print((event.message.offset, event.value))

票数 0

页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持

原文链接：

https://stackoverflow.com/questions/60267358

复制

相似问题

问如何将Faust中的消费者设置为特定的偏移量
EN

回答 2

Stack Overflow用户

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何将Faust中的消费者设置为特定的偏移量EN

回答 2

Stack Overflow用户

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何将Faust中的消费者设置为特定的偏移量
EN