python流数据动态可视化

Streaming Data

“流数据”是连续生成的数据,通常由某些外部源(如远程网站,测量设备或模拟器)生成。这种数据在金融时间序列,Web服务器日志,科学应用程序和许多其他情况下很常见。我们已经了解了如何在[实时数据](06-Live _Data.ipynb)用户指南中显示可调用的任何数据输出,我们还看到了如何使用HoloViews流系统在用户指南中推送事件部分[响应事件](11-响应_到Events.ipynb)和[自定义交互](12-Custom Interactivity.ipynb)。

本用户指南显示了使用DynamicMap和流构建交互式绘图的第三种方法。在这里,不是将绘图元数据(例如缩放范围,用户触发的事件,如“Tap”等)推送到DynamicMap回调,而是使用HoloViews直接更新可视化元素中的基础数据。 `Stream``。

特别是,我们将展示如何使用HoloViews的PipeBuffer流来处理流数据源,而无需从DynamicMap可调用内部获取或生成数据。除了简单地从“DynamicMap”外部设置元素数据之外,我们还将探索使用可选单独[streamz]协调的流数据处理方法 (http://matthewrocklin.com/blog/work/来自MattRocklin的2017/10/16/streaming-dataframes-1) 库可以使构建复杂的流媒体管道变得更加简单。

请注意,此页面演示了需要实时运行的Python服务器的功能。当导出到pyviz.org网站上的静态HTML页面时,您将只看到一个图。当将此代码作为Jupyter笔记本运行时,您应该逐个单元地执行它以依次查看每个操作的效果。

In [ ]:

import time
import numpy as np
import pandas as pd
import holoviews as hv

from holoviews.streams import Pipe, Buffer

import streamz
import streamz.dataframe

hv.extension('bokeh')

Pipe

Pipe允许将数据推送到DynamicMap回调中以更改可视化,就像[响应事件](./ 11-Responding to Events.ipynb)用户指南中的流一样用于将更改推送到控制可视化的元数据。可以使用Pipe来推送任何类型的数据,并使其可用于DynamicMap回调。由于所有Element类型接受各种形式的data,我们可以使用Pipe通过DynamicMap将数据直接推送到Element的构造函数。

我们可以利用大多数元素可以在不提供任何数据的情况下实例化的事实,因此我们使用空列表声明Pipe并声明DynamicMap,将管道作为流提供,这将动态更新VectorField

In [ ]:

pipe = Pipe([])
vector_dmap = hv.DynamicMap(hv.VectorField, streams=[pipe])
vector_dmap.redim.range(x=(-1, 1), y=(-1, 1))

设置这个VectorField绑定到Pipe后,我们可以开始向它推送数据,改变VectorField的方向:

In [ ]:

x,y  = np.mgrid[-10:11,-10:11] * 0.1
sine_rings  = np.sin(x**2+y**2)*np.pi+np.pi
exp_falloff = 1/np.exp((x**2+y**2)/8)

for i in np.linspace(0, 1, 25):
    time.sleep(0.1)
    pipe.send([x,y,sine_rings*i, exp_falloff])

这种直接使用元素构造函数的方法不允许您使用除默认键和值维之外的任何内容。这个限制的一个简单的解决方法是使用functools.partial,如下面控制长度部分中所示。

由于Pipe是完全通用的,数据可以是任何自定义类型,因此它提供了一种完整的通用机制来传输结构化或非结构化数据。由于这种普遍性,Pipe在使用下一节中描述的Buffer流时不提供一些更复杂的功能和优化。

Buffer

虽然Pipe提供了将任意数据传递给DynamicMap回调的通用解决方案,但另一方面Buffer提供了一种非常强大的方法来处理流表格数据,定义为pandas数据帧,数组,或列的词典(以及StreamingDataFrame,稍后我们将介绍)。 Buffer自动累积表格数据的最后一行N行,其中Nlength定义。

累积数据的能力允许对最近的数据历史执行操作,而绘制后端(例如散景)可以通过仅发送最新的补丁来优化绘图更新。只有当Buffer持有的data对象与绘制的Element数据相同时,此优化才有效,否则所有数据都将正常更新。

一个简单的例子:布朗运动

要初始化Buffer,我们必须提供一个示例数据集,它定义我们将要流式传输的数据的列和dtypes。接下来,我们定义length以保留最后100行数据。如果数据是DataFrame,我们可以指定是否还要使用DataFrame````index。在这种情况下,我们将简单地定义我们想要绘制'x'和'y'位置的DataFrame和'count'作为PointsCurve元素:

In [ ]:

example = pd.DataFrame({'x': [], 'y': [], 'count': []}, columns=['x', 'y', 'count'])
dfstream = Buffer(example, length=100, index=False)
curve_dmap = hv.DynamicMap(hv.Curve, streams=[dfstream])
point_dmap = hv.DynamicMap(hv.Points, streams=[dfstream])

应用了一些样式后,我们将显示动态“曲线”和“点”的“叠加”。

In [ ]:

%%opts Points [color_index='count', xaxis=None, yaxis=None] (line_color='black', size=5)
%%opts Curve (line_width=1, color='black')
curve_dmap * point_dmap

现在我们已经设置了Buffer并定义了一个DynamicMap来绘制数据,我们可以开始向它推送数据。我们将定义一个简单的函数,通过累积x,y位置来模拟布朗运动。我们可以通过hv.streams.Buffer直接“发送”`数据。

In [ ]:

def gen_brownian():
    x, y, count = 0, 0, 0
    while True:
        x += np.random.randn()
        y += np.random.randn()
        count += 1
        yield pd.DataFrame([(x, y, count)], columns=['x', 'y', 'count'])

brownian = gen_brownian()
for i in range(200):
    dfstream.send(next(brownian))

最后,我们可以使用clear方法清除流和绘图上的数据:

In [ ]:

#dfstream.clear()

使用Streamz库

现在我们已经发现了什么PipeBuffer可以做它的时间来展示如何将它们与streamz库一起使用。虽然HoloViews不依赖于streamz并且您可以使用流功能而无需了解streamz,但这两个库可以很好地协同工作,允许您构建管道来管理连续的数据流。 Streamz易于用于简单的任务,但也支持复杂的管道,涉及分支,连接,流量控制,反馈等。这里我们主要关注将streamz输出连接到Pipe然后Buffer以获得有关streamz API的更多细节,请参阅[streamz文档](https://streamz.readthedocs.io/en/最新/)。

streamz.StreamPipe一起使用

让我们从一个相当简单的例子开始:

  1. 声明一个streamz.Stream和一个Pipe对象,并将它们连接到一个我们可以推送数据的管道中。
  2. 使用20的sliding_window,它将首先等待20组流更新累积。此时,对于每个后续更新,它将应用pd.concat将最近的20个更新组合到一个新的数据帧中。
  3. 使用streamz.Stream上的sink方法来send得到20个更新为Pipe的集合。
  4. 声明一个DynamicMap,它采用连接的DataFrames的滑动窗口,并使用Scatter元素显示它。
  5. 用“计数”对“散点图”点进行着色并设置范围,然后显示:

In [ ]:

point_source = streamz.Stream()
pipe = Pipe(data=[])
point_source.sliding_window(20).map(pd.concat).sink(pipe.send) # Connect streamz to the Pipe
scatter_dmap = hv.DynamicMap(hv.Scatter, streams=[pipe])

设置完流媒体管道后,我们可以再次显示它:

In [ ]:

%%opts Scatter [color_index='count', bgcolor='black']
scatter_dmap.redim.range(y=(-4, 4))

现在有一个管道,但最初这个图是空的,因为没有数据发送到它。要查看情节更新,让我们使用streamz.Streamemit方法将小块随机大熊猫DataFrames发送到我们的情节:

In [ ]:

for i in range(100):
    df = pd.DataFrame({'x': np.random.rand(100), 'y': np.random.randn(100), 'count': i},
                      columns=['x', 'y', 'count'])
    point_source.emit(df)

使用StreamingDataFrame和StreamingSeries

streamz库(版本0.2)提供了StreamingDataFrameStreamingSeries,它是一种轻松处理表格数据实时源的有效方法。这使得它非常适合使用Buffer。 (在版本0.3streamz已经重命名了类来删除Streaming,所以这里的代码需要更新以用于后来的streamz版本。)使用StreamingDataFrame我们可以轻松传输数据,应用累积和滚动统计等计算,然后使用HoloViews可视化数据。

streamz.dataframe模块提供了一个Random实用程序,它生成一个StreamingDataFrame,它以指定的间隔发出一定频率的随机数据。 example属性让我们可以看到我们可以期待的数据的结构和dtypes:

In [ ]:

simple_sdf = streamz.dataframe.Random(freq='10ms', interval='100ms')
print(simple_sdf.index)
simple_sdf.example.dtypes

由于StreamingDataFrame提供了类似pandas的API,我们可以直接指定对数据的操作。在这个例子中,我们减去一个固定的偏移,然后计算累积和,给我们一个随机漂移的时间序列。然后我们可以将这个数据帧的x值传递给HoloViews的Buffer并提供hv.Curve作为DynamicMap回调,将数据流式传输到HoloViewsCurve(带有默认键和值维度):

In [ ]:

%%opts Curve [width=500 show_grid=True]
sdf1 = (simple_sdf-0.5).cumsum()
hv.DynamicMap(hv.Curve, streams=[Buffer(sdf1.x)])

Random StreamingDataFrame将异步发出事件,向前驱动可视化,直到它被显式停止,我们可以通过调用stop方法来做。

In [ ]:

simple_sdf.stop()

使用StreamingDataFrame API

到目前为止,我们只计算了累积和,但是StreamingDataFrame实际上有一个广泛的API,可以让我们对我们的数据进行广泛的流式计算。例如,让我们将滚动均值应用于我们的x值,窗口为500毫秒,并将其叠加在“原始”数据之上:

In [ ]:

source_df = streamz.dataframe.Random(freq='5ms', interval='100ms')
sdf2 = (source_df-0.5).cumsum()
sdf2.tail()

In [ ]:

raw       = hv.DynamicMap(hv.Curve, streams=[Buffer(sdf2.x)]).relabel('raw')
smooth    = hv.DynamicMap(hv.Curve, streams=[Buffer(sdf2.x.rolling('500ms').mean())]).relabel('smooth')

raw * smooth.options(width=900,height=350)

In [ ]:

source_df.stop()

对流数据的操作

正如我们在上面发现的那样,Buffer让我们设置一个length,它定义了我们想要累积的行数。我们可以利用这个优势并在此长度窗口上应用操作。在这个例子中,我们声明一个Dataset然后应用histogram操作来计算指定length窗口上的Histogram

In [ ]:

hist_source = streamz.dataframe.Random(freq='5ms', interval='100ms')
sdf4 = (hist_source-0.5).cumsum()

dmap = hv.DynamicMap(hv.Dataset, streams=[Buffer(sdf4.x, length=500)])
hv.operation.histogram(dmap, dimension='x')

In [ ]:

hist_source.stop()

Datashading

同样的方法也适用于数据分析器操作,即使我们使它非常大(在这种情况下为100万个样本),也让我们对整个“长度”窗口进行数据分析:

In [ ]:

%%opts RGB [width=600]
from holoviews.operation.datashader import datashade
from bokeh.palettes import Blues8
large_source = streamz.dataframe.Random(freq='100us', interval='200ms')
sdf5 = (large_source-0.5).cumsum()

dmap = hv.DynamicMap(hv.Curve, streams=[Buffer(sdf5.x, length=1000000)])
datashade(dmap, streams=[hv.streams.PlotSize], normalization='linear', cmap=Blues8)

In [ ]:

large_source.stop()

使用tornadoIOLoop进行异步更新

在大多数情况下,您不希望在同一个Python进程中手动推送更新,而是希望对象在新数据到达时异步更新。由于Jupyter和Bokeh服务器都在[tornado](http://www.tornadoweb.org/en/stable/)上运行,我们可以在两种情况下使用tornado``IOLoop``来定义非阻塞协同例程这可以在数据准备就绪时将数据推送到我们的流中。

In [ ]:

%%opts Curve [width=600]
from tornado.ioloop import PeriodicCallback

count = 0
buffer = Buffer(np.zeros((0, 2)), length=50)

def f():
    global count
    count += 1
    buffer.send(np.array([[count, np.random.rand()]]))
    
cb = PeriodicCallback(f, 100)
cb.start()

hv.DynamicMap(hv.Curve, streams=[buffer]).redim.range(y=(0, 1))

In [ ]:

cb.stop()

在这里,当您执行笔记本电脑时,图表应该像以前一样更新100次,但现在通过Tornado IOLoop不会阻止其他交互并在笔记本中工作。

真实的例子

使用PipeBuffer流我们可以非常容易地创建复杂的流图。除了我们在本指南中介绍的玩具示例之外,还有必要查看使用真实,实时,流数据的一些示例。

  • [streaming psutil](http://holoviews.org/gallery/apps/bokeh/stream psutil.html)散景应用程序是一个这样的例子,使用psutil`显示CPU和内存信息`library(用pip install psutilconda install psutil``安装)

如您所见,流数据通常像HoloViews中的流一样工作,在显式控制下灵活处理随时间变化或由某些外部数据源控制。

本教程的最后几节将介绍如何将目前为止所涉及的所有概念纳入交互式Web应用程序以处理大型或小型数据集,首先介绍[参数和小部件](./ 12 <_>参数<_> and_Widgets.ipynb)。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏犀利豆的技术空间

Java 渲染 docx 文件,并生成 pdf 加水印

一顿google以后发现了 StackOverflow 上的这个回答:Converting docx into pdf in java 使用如下的 jar 包:

47410
来自专栏乐享123

How to Sort a Very Very Very Big File

17250
来自专栏全栈之路

关于Canvas保存为图片

但是在webapp该方法是不行的,默认是不支持的。一种方法是在android的java代码写js接口,而一个纯webapp,确是很难做到(其实也不是),只不过找...

1.1K10
来自专栏Windows Community

Windows Community Toolkit 3.0 - Gaze Interaction

Gaze Input & Tracking - 也就是视觉输入和跟踪,是一种和鼠标/触摸屏输入非常不一样的交互方式,利用人类眼球的识别和眼球方向角度的跟踪,来判...

13430
来自专栏数据小魔方

批量将本地gis数据导入postgis数据库

以前在处理gis数据的时候,都是直接导入本地shp素材、本地geojson素材,本地topojson素材,自从接触postgis数据之后,深感使用规范的存储系统...

21710
来自专栏北京马哥教育

shell十三问,为linux学习打基础(一)

本文整理并转自CU上的帖子[学习共享] shell 十三問?,此贴是2003年发表的,但却是相当不错的linux基础知识汇集贴,原帖主使用的台湾风格,本文加以简...

39640
来自专栏zhisheng

Java研发方向如何准备BAT技术面试答案(下)

本文是针对文章《 Java研发方向如何准备BAT技术面试(超级干货)》里面的算法、数据结构、Linux和操作系统问题的一些答案。如有错误,还请各位网友指正。多谢...

1.1K340
来自专栏听雨堂

【3】利用Word模板生成文档的总结

阅读目录 Word二次开发概况 使用DsoFramer进行开发 使用Interop进行开发 打开、关闭和写入操作 批量替换文本 遍历段落替换文本 查找后逐个替换...

38750
来自专栏happyJared

Elasticsearch地理坐标类型(Geo-point)在Spring Data ES中的常见使用问题整理解答

  下文整理的几个问答,本人在实际应用中亲身经历或解决过的,主要涉及Elasticsearch地理坐标类型(Geo-point)在Java应用中的一些特殊使用场...

45410
来自专栏Spark学习技巧

Rowkey(行键)设计

HBase 中的行按行键按顺序排序。这种设计优化了扫描(scan),允许您将相关的行或彼此靠近的行一起读取。但是,设计不佳的行键是 hotspotting 的常...

17420

扫码关注云+社区

领取腾讯云代金券