如何使用 Holoviews 将数据流式传输到 Bokeh 服务器答案

【问题标题】：How to stream data to a Bokeh server using Holoviews如何使用 Holoviews 将数据流式传输到 Bokeh 服务器
【发布时间】：2019-06-11 11:38:28
【问题描述】：

我想将 python 生成的数据流式传输到网页。我想出了以下示例，使用来自
https://holoviews.org/user_guide/Streaming_Data.html
的示例和
http://holoviews.org/user_guide/Deploying_Bokeh_Apps.html
但是我收到一个文档锁定错误：
"'_pending_writes 应该是 non-None 当我们有文档锁时，我们应该有锁当文档改变时'"

这是我的例子：

import numpy as np
import holoviews as hv
import holoviews.plotting.bokeh
import streamz
import streamz.dataframe
renderer = hv.renderer('bokeh')
from holoviews import opts
from holoviews.streams import Pipe, Buffer
hv.extension('bokeh')


source_df = streamz.dataframe.Random(freq='5ms', interval='100ms')
sdf = (source_df-0.5).cumsum()
raw_dmap = hv.DynamicMap(hv.Curve, streams=[Buffer(sdf.x)])
smooth_dmap = hv.DynamicMap(hv.Curve, streams=[Buffer(sdf.x.rolling('50ms').mean())])

fig = (raw_dmap.relabel('raw') * smooth_dmap.relabel('smooth')).opts(
    opts.Curve(width=500, show_grid=True))

server = renderer.app(fig, show=True, new_window=True)

一个页面打开，图形显示但没有更新。在我的笔记本中，我收到以下错误：

tornado.application - ERROR - Exception in callback functools.partial(<function wrap.<locals>.null_wrapper at 0x00000234E3CB9400>, <Future finished exception=RuntimeError('_pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes')>)
Traceback (most recent call last):
  File "C:\ProgramData\Anaconda3\lib\site-packages\tornado\ioloop.py", line 758, in _run_callback
    ret = callback()
  File "C:\ProgramData\Anaconda3\lib\site-packages\tornado\stack_context.py", line 300, in null_wrapper
    return fn(*args, **kwargs)
  File "C:\ProgramData\Anaconda3\lib\site-packages\tornado\ioloop.py", line 779, in _discard_future_result
    future.result()
  File "C:\ProgramData\Anaconda3\lib\site-packages\tornado\gen.py", line 1147, in run
    yielded = self.gen.send(value)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\dataframe\core.py", line 802, in _cb
    yield source._emit((last, now, freq))
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 298, in _emit
    r = downstream.update(x, who=self)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 563, in update
    return self._emit(result)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 298, in _emit
    r = downstream.update(x, who=self)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 563, in update
    return self._emit(result)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 298, in _emit
    r = downstream.update(x, who=self)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 747, in update
    return self._emit(result)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 298, in _emit
    r = downstream.update(x, who=self)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 563, in update
    return self._emit(result)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 298, in _emit
    r = downstream.update(x, who=self)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 563, in update
    return self._emit(result)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 298, in _emit
    r = downstream.update(x, who=self)
  File "C:\ProgramData\Anaconda3\lib\site-packages\streamz\core.py", line 516, in update
    result = self.func(x, *self.args, **self.kwargs)
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\streams.py", line 436, in send
    self.event(data=data)
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\streams.py", line 375, in event
    self.trigger([self])
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\streams.py", line 156, in trigger
    subscriber(**dict(union))
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\plotting\plot.py", line 615, in refresh
    self._trigger_refresh(stream_key)
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\plotting\plot.py", line 624, in _trigger_refresh
    self.update(key)
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\plotting\plot.py", line 596, in update
    item = self.__getitem__(key)
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\plotting\plot.py", line 261, in __getitem__
    self.update_frame(frame)
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\plotting\bokeh\element.py", line 1944, in update_frame
    self._update_ranges(element, ranges)
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\plotting\bokeh\element.py", line 657, in _update_ranges
    self._shared['x'], self.logx, streaming)
  File "C:\ProgramData\Anaconda3\lib\site-packages\holoviews\plotting\bokeh\element.py", line 702, in _update_range
    axis_range.trigger(k, old, new)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\model.py", line 599, in trigger
    super(Model, self).trigger(attr, old, new, hint=hint, setter=setter)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\util\callback_manager.py", line 143, in trigger
    self._document._notify_change(self, attr, old, new, hint, setter, invoke)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 1004, in _notify_change
    self._trigger_on_change(event)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 1099, in _trigger_on_change
    self._with_self_as_curdoc(invoke_callbacks)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 1112, in _with_self_as_curdoc
    return f()
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 1098, in invoke_callbacks
    cb(event)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\document.py", line 668, in <lambda>
    self._callbacks[receiver] = lambda event: event.dispatch(receiver)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\events.py", line 244, in dispatch
    super(ModelChangedEvent, self).dispatch(receiver)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\document\events.py", line 126, in dispatch
    receiver._document_patched(self)
  File "C:\ProgramData\Anaconda3\lib\site-packages\bokeh\server\session.py", line 214, in _document_patched
    raise RuntimeError("_pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes")
RuntimeError: _pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes

任何线索我做错了什么？亲切的问候

【问题讨论】：

Bokeh 期望其对象的更新发生在“锁定”回调中，例如在其通常的on_change 或周期性或“下一个滴答”回调中。当某些东西试图规避该要求时会发生此错误消息，例如通过在任何 Bokeh 回调之外的另一个线程直接更新 Bokeh 对象。几乎可以肯定，它表明 streamz 或 Holoviews 中存在使用错误。

标签： python-3.x server streaming bokeh holoviews

【解决方案1】：

我将最后一行更改为renderer.server_doc(fig)，将所有内容保存为名为 test.ipynb 的笔记本。在命令提示符下，我运行了>bokeh serve --show .\test.ipynb。服务器已启动，数据流按预期显示在浏览器中。

import numpy as np
import holoviews as hv
import holoviews.plotting.bokeh
import streamz
import streamz.dataframe
renderer = hv.renderer('bokeh')
from holoviews import opts
from holoviews.streams import Pipe, Buffer
hv.extension('bokeh')


source_df = streamz.dataframe.Random(freq='5ms', interval='100ms')
sdf = (source_df-0.5).cumsum()
raw_dmap = hv.DynamicMap(hv.Curve, streams=[Buffer(sdf.x)])
smooth_dmap = hv.DynamicMap(hv.Curve, streams=[Buffer(sdf.x.rolling('50ms').mean())])

fig = (raw_dmap.relabel('raw') * smooth_dmap.relabel('smooth')).opts(
    opts.Curve(width=500, show_grid=True))

renderer.server_doc(fig)

【讨论】：