如何在多个代理或浮士德计时器之间共享浮士德表？答案

【问题标题】：How to share faust table between multiple agents or faust timers?如何在多个代理或浮士德计时器之间共享浮士德表？
【发布时间】：2020-01-25 12:03:54
【问题描述】：

我正在尝试在一段时间后将浮士德表的数据（计数）发布到 kafka 主题。当我发布一些简单的字符串时，计时器正在工作，但它无法以某种方式访问表的数据。下面是定时器的代码：

@app.timer(interval=10.0)
async def publish_to_anomaly_topic():
            await anomaly_topic.send(
            value=str(page_views['total'].value())
          )
@app.agent(page_view_topic)
async def count_page_views(views):
    async for view in views.group_by(PageView.id):
        total=0
        page_views[view.id]+=1
        for everykey in list(page_views.keys()):
            if everykey != 'total':
                total+=page_views[everykey].value()
        page_views['total'] = total

代理工作正常。我能够正确地看到这些值。

【问题讨论】：

您能分享一下您遇到的错误或故障吗？只是为了了解浮士德的机制。非常感谢

标签： python-3.x apache-kafka faust

【解决方案1】：

在我尝试做同样的事情时发现了这个问题，这就是我能够弄清楚的方法。

https://faust.readthedocs.io/en/latest/userguide/tables.html

您不能在流操作之外修改表；这意味着您只能在异步事件中更改表 stream: block. 我们需要它来对齐表的分区流的，并确保源主题分区正确在失败时重新平衡到不同的工人，以及任何必要的表分区。

在流之外修改表会引发错误：

文档说您无法在流操作之外访问/修改表。

要解决这个问题，您可以将计时器功能分成两部分：

@app.timer(10)
async def my_timer_function():
    # value does not matter as much as the send operation
    await my_calling_function.send(value="send data now!") 

@app.agent()
async def my_calling_function(stream_from_timer_func):
    async for message in stream_from_timer_func:
        print(message) # this will print "send data now!"
        table_data = my_table['key']
        # Here is where you can access your table data and finish sending the 
        # message to the topic you want
        await my_topic.send(value=table_data)

如您所见，如果您使用计时器功能向代理发送消息，则可以访问所需的表，它只需要在一个

async for event in stream:

代码块。

【讨论】：

【解决方案2】：

经过大量实验，事实证明您无法访问表的值以及应用程序计时器（即使您在创建表时指定了relative_field 选项）。此问题的解决方法是创建另一个表来维护您的消息的时间戳并在您的业务逻辑中使用它们。

   if view.timestamp-page_views_timer[view.id+'_first_timestamp'] > 60:
         await anomaly_topic.send(value={//the data to be sent})

其中 page_views_timer 是创建的新表。

【讨论】：