Plotly Dash/Apache Kafka - 我希望看到一个实时绘图而不是在完成绘图时出现答案

【问题标题】：Plotly Dash/ Apache Kafka - I would like to see a graph plotting live rather than appearing when finished plottingPlotly Dash/Apache Kafka - 我希望看到一个实时绘图而不是在完成绘图时出现
【发布时间】：2021-07-06 17:01:02
【问题描述】：

我已经在 plotly 中查看了流图的实时更新，但这并不是我想要做的。我有一个图形组件，单击按钮组件即可填充该组件。在更新图形的回调中，创建了一个对象，并从该类调用了一个方法，该方法使用 apache kafka 流生产者返回一个 15 个值的流。然后，我在回调中有 apache kafka 消费者，它一一接收值。但是，我试图让这些值也一一出现在我的仪表板上。我尝试使用间隔组件，但这似乎只是每秒调用一次函数，这意味着脚本在调用函数时会卡住并且不执行任何操作。

如果有任何帮助，我们将不胜感激，如有必要，可以提供更多详细信息。

【问题讨论】：

您好，更多细节会很棒，可能是代码 sn-p 。但是从我读到的内容中，您想要执行回调-> pub/sub 函数而不是间隔，一旦您从kafka 获得事件，您想要将事件异步推送到侦听器，这将确保它不会阻塞.. 但是有一个 sn-p 我可以提供进一步的帮助
@Gabriel 我已经包含了 sn-p
问几个问题，你们是怎么重构消费者点击的？，我要做的是在构造函数的 init 上构造它们..所以就在 some_object.simulate() 下面那么你也应该使用 multiprocessing.Process 线程化消费者，就像你对生产者所做的那样，然后当消费者触发它应该更新列表上的数据，然后当您的间隔命中时，您只需从该列表中读取，或者如果您有智能绑定，您可以直接推送它 - 这有意义吗？如果仍然没有运气，请尝试一下我可以发布我的解决方案 - 流数据始终处于打开状态和后台
感谢您的回复@Gabriel，我试过了，但我仍然遇到在模拟和填充图形函数中创建对象的问题，它似乎试图创建对象 6 次，但我只希望它创建一次
没问题，消费者不应该在回调函数中创建。我将为您做一个工作示例并将其与一些 cmets 一起发布。明天再来看看:)

标签： python apache-kafka plotly-dash

【解决方案1】：

使用 Streams，您将希望将使用者置于其自己的线程中。您已经将生产者放在线程中，我也建议不要将生产者加入主线程

您想要做的是将消费者放在一个线程中并使用队列或线程安全数据结构或锁定，队列是线程安全的，因此您可以根据需要提取数据

您还希望在 def 或 startup 中运行线程，而不是在回调中运行，每次在回调中，您只会询问队列中的最新项目，并且可以将间隔设置为 100 毫秒左右 dep on pref

下面有一个简单的例子：

while 循环将是您的回调间隔 - 我使用 while 进行时间
您必须将使用者设置在回调之外 - 否则您将一次又一次地构建它，将使用者置于回调范围之外并在启动时运行它，下面我使用了 main。


    from time import sleep
    from json import dumps
    from kafka import KafkaConsumer
    from multiprocessing import Process, Queue
    import os
    
    
    
    def consumer(q):
    
        consumer_obj = KafkaConsumer('test')
        for message in  consumer_obj:
         q.put(message.value)
    
    
    if __name__ == '__main__':
        q = Queue()
        p = Process(target=consumer, args=(q,))
        p.start()
        while True:
            print(q.get())
            sleep(1)

【讨论】：

【解决方案2】：

消费者循环是无限的，您不能为同一组记录重复相同的消费者对象两次。您不能使用列表理解，因此您需要单独存储列表

以下内容未经测试，但显示了您需要做什么的大致思路

x = []
y = []

LIMIT = 15

for r in consumer_obj:

  if len(x) >= LIMIT:
    del x[0]
  if len(y) >= LIMIT:
    del y[0]

  x.append(r.value['day'])
  y.append(r.value['biomass'])
  
  # TODO: Update graph 
  data = {"data": [
          {"x": x,"y": y, 
          "type": "lines",
        }]

这里有一个项目使用 Bokeh int 而不是 Plotly - https://github.com/Aakash282/kafka-bokeh-dashboard

或者你可以只使用 Kafka Connect 写入 SQLite 数据库，并让你的情节使用它并定期更新

【讨论】：