preface

流式數(shù)據(jù)的監(jiān)控，以下主要是從算法的呈現(xiàn)出發(fā)，提供一種python的實(shí)現(xiàn)思路

其中：

1.python是2.X版本

2.提供兩種實(shí)現(xiàn)思路，一是基于matplotlib的animation，一是基于matplotlib的ion

話不多說(shuō)，先了解大概的效果，如下：

一、一點(diǎn)構(gòu)思

在做此流數(shù)據(jù)輸出可視化前，一直在搗鼓nupic框架，其內(nèi)部HTM算法主要是一種智能的異常檢測(cè)算法，是目前AI框架中垂直領(lǐng)域下的一股清流，但由于其實(shí)現(xiàn)的例子對(duì)應(yīng)的流數(shù)據(jù)展示并非我想要的，故此借鑒后自己重新寫(xiě)了一個(gè)，主要是達(dá)到三個(gè)目的，一是展示真實(shí)數(shù)據(jù)的波動(dòng)，二是展示各波動(dòng)的異常得分，三是羅列異常的點(diǎn)。

上述的輸出結(jié)構(gòu)并非重點(diǎn)，重點(diǎn)是其實(shí)時(shí)更新的機(jī)制，了解后即可自行定義。另，js對(duì)于這種流數(shù)據(jù)展示應(yīng)該不難，所以本文主要立足的是算法的呈現(xiàn)角度以及python的實(shí)現(xiàn)。

二、matplotlib animation實(shí)現(xiàn)思路

（一）、骨架與實(shí)時(shí)更新

animation翻譯過(guò)來(lái)就是動(dòng)畫(huà)，其動(dòng)畫(huà)展示核心主要有三個(gè)：1是動(dòng)畫(huà)的骨架先搭好，就是圖像的邊邊框框這些，2是更新的過(guò)程，即傳入實(shí)時(shí)數(shù)據(jù)時(shí)圖形的變化方法，3是FuncAnimation方法結(jié)尾。

下面以一個(gè)小例子做進(jìn)一步說(shuō)明：

1.對(duì)于動(dòng)畫(huà)的骨架：

# initial the figure.

x = []

y = []

fig = plt.figure(figsize=(18, 8), facecolor="white")

ax1 = fig.add_subplot(111)

p1, = ax1.plot(x, y, linestyle="dashed", color="red")

以上分別對(duì)應(yīng)初始化空數(shù)據(jù)，初始化圖形大小和背景顏色，插入子圖（三個(gè)數(shù)字分別表示幾行幾列第幾個(gè)位置），初始化圖形（數(shù)據(jù)為空）。

import numpy as np

x = np.arange(0, 1000, 1)

y = np.random.normal(100, 10, 1000)

隨機(jī)生成一些作圖數(shù)據(jù)，下面定義update過(guò)程。

2.對(duì)于更新過(guò)程：

def update(i):

x.append(xs[i])

y.append(ys[i])

ax1.set_xlim(min(x),max(x)+1)

ax1.set_ylim(min(y),max(y)+1)

p1.set_data(x,y)

ax1.figure.canvas.draw()

return p1

上述定義更新函數(shù)，參數(shù)i為每輪迭代從FuncAnimation方法frames參數(shù)傳進(jìn)來(lái)的數(shù)值，frames參數(shù)的指定下文會(huì)進(jìn)一步說(shuō)，x/y通過(guò)相應(yīng)更新之后，對(duì)圖形的x/y軸大小做相應(yīng)的重設(shè)，再把數(shù)據(jù)通過(guò)set_data傳進(jìn)圖形，注意ax1和p1的區(qū)別，最后再把上述的變化通過(guò)draw()方法繪制到界面上，返回p1給FuncAnimation方法。

3.對(duì)于FuncAnimation方法：

ani = FuncAnimation(fig=fig,func=update,frames=len(xs),interval=1)

plt.show()

FuncAnimation方法主要是與update函數(shù)做交互，將frames參數(shù)對(duì)應(yīng)的數(shù)據(jù)逐條傳進(jìn)update函數(shù)，再由update函數(shù)返回的圖形覆蓋FuncAnimation原先的圖形，fig參數(shù)即為一開(kāi)始對(duì)應(yīng)的參數(shù)，interval為每次更新的時(shí)間間隔，還有其他一些參數(shù)如blit=True控制圖形精細(xì)，當(dāng)界面較多子圖時(shí)，為T(mén)rue可以使得看起來(lái)不會(huì)太卡，關(guān)鍵是frames參數(shù)，下面是官方給出的注釋：

可為迭代數(shù)，可為函數(shù)，也可為空，上面我指定為數(shù)組的長(zhǎng)度，其迭代則從0開(kāi)始到最后該數(shù)值停止。

該例子最終呈現(xiàn)的效果如下：

了解大概的實(shí)現(xiàn)，細(xì)節(jié)就不在這里多說(shuō)了。

（二）、animation的優(yōu)缺點(diǎn)

animation的繪制的結(jié)果相比于下文的ion會(huì)更加的細(xì)膩，主要體現(xiàn)在FuncAnimation方法的一些參數(shù)的控制上。但是缺點(diǎn)也是明顯，就是必須先有指定的數(shù)據(jù)或者指定的數(shù)據(jù)大小，顯然這樣對(duì)于預(yù)先無(wú)法知道數(shù)據(jù)的情況沒(méi)法處理。所以換一種思路，在matplotlib ion打開(kāi)的模式下，每次往模板插入數(shù)據(jù)都會(huì)進(jìn)行相應(yīng)的更新，具體看第二部分。

三、matplotlib ion實(shí)現(xiàn)思路

（一）、實(shí)時(shí)更新

matplotlib ion的實(shí)現(xiàn)也主要是三個(gè)核心，1是打開(kāi)ion，2是實(shí)時(shí)更新機(jī)制，3是呈現(xiàn)在界面上。

1.對(duì)于打開(kāi)ion：

ion全稱(chēng)是 interactive on（交互打開(kāi)），其意為打開(kāi)一個(gè)圖形的交互接口，之后每次繪圖都在之前打開(kāi)的面板上操作，舉個(gè)例子：

import matplotlib.pyplot as plt

plt.ion()

fig = plt.figure()

ax1 = fig.add_subplot(111)

line, = ax1.plot(t, v, linestyle="-", color="r")

打開(kāi)交互接口，初始化圖形。

2.對(duì)于實(shí)時(shí)更新機(jī)制：

import numpy as np

ys = np.random.normal(100, 10, 1000)

def p(a, b):

t.append(a)

v.append(b)

ax1.set_xlim(min(t), max(t) + 1)

ax1.set_ylim(min(v), max(v) + 1)

line.set_data(t, v)

plt.pause(0.001)

ax1.figure.canvas.draw()

for i in xrange(len(ys)):

p(i, ys[i])

隨機(jī)生成一組數(shù)據(jù)，定義作圖函數(shù)p（包含pause表示暫定時(shí)延，最好有，防止界面卡死），傳入數(shù)據(jù)實(shí)時(shí)更新。

3.對(duì)于界面最終呈現(xiàn)

plt.ioff()

plt.show()

ioff是關(guān)閉交互模式，就像open打開(kāi)文件產(chǎn)生的句柄，最好也有個(gè)close關(guān)掉。

最終效果如下：

（二）、ion的優(yōu)缺點(diǎn)

animation可以在細(xì)節(jié)上控制比ion更加細(xì)膩，這也是ion沒(méi)有的一點(diǎn)，但是單就無(wú)需預(yù)先指定數(shù)據(jù)這一點(diǎn)，ion也無(wú)疑是能把流數(shù)據(jù)做得更加好。

四、最后

貼一下兩種方法在最開(kāi)始那種圖的做法，ion我定義成類(lèi)，這樣每次調(diào)用只需穿入?yún)?shù)就可以。

animation版本

# _*_ coding:utf-8 _*_

import os

import csv

import datetime

import matplotlib.pyplot as plt

from matplotlib.animation import FuncAnimation

from matplotlib.dates import DateFormatter

import matplotlib.ticker as ticker

# read the file

filePath = os.path.join(os.getcwd(), "data/anomalyDetect_output.csv")

file = open(filePath, "r")

allData = csv.reader(file)

# skip the first three columns

allData.next()

allData.next()

allData.next()

# cache the data

data = [line for line in allData]

# for i in data: print i

# take out the target value

timestamp = [line[0] for line in data]

value = [line[1:] for line in data]

# format the time style 2016-12-01 00:00:00

def timestampFormat(t):

result = datetime.datetime.strptime(t, "%Y-%m-%d %H:%M:%S")

return result

# take out the data

timestamp = map(timestampFormat, timestamp)

value_a = [float(x[0]) for x in value]

predict_a = [float(x[1]) for x in value]

anomalyScore_a = [float(x[2]) for x in value]

# initial the size of the figure

fig = plt.figure(figsize=(18, 8), facecolor="white")

fig.subplots_adjust(left=0.06, right=0.70)

ax1 = fig.add_subplot(2, 1, 1)

ax2 = fig.add_subplot(2, 1, 2)

ax3 = fig.add_axes([0.8, 0.1, 0.2, 0.8], frameon=False)

# initial plot

p1, = ax1.plot_date([], [], fmt="-", color="red", label="actual")

ax1.legend(loc="upper right", frameon=False)

ax1.grid(True)

p2, = ax2.plot_date([], [], fmt="-", color="red", label="anomaly score")

ax2.legend(loc="upper right", frameon=False)

ax2.axhline(0.8, color='black', lw=2)

# add the x/y label

ax2.set_xlabel("date time")

ax2.set_ylabel("anomaly score")

ax1.set_ylabel("value")

# add the table in ax3

col_labels = ["date time", 'actual value', 'predict value', 'anomaly score']

ax3.text(0.05, 0.99, "anomaly value table", size=12)

ax3.set_xticks([])

ax3.set_yticks([])

# axis format

dateFormat = DateFormatter("%m/%d %H:%M")

ax1.xaxis.set_major_formatter(ticker.FuncFormatter(dateFormat))

ax2.xaxis.set_major_formatter(ticker.FuncFormatter(dateFormat))

# define the initial function

def init():

p1.set_data([], [])

p2.set_data([], [])

return p1, p2

# initial data for the update function

x1 = []

x2 = []

x1_2 = []

y1_2 = []

x1_3 = []

y1_3 = []

y1 = []

y2 = []

highlightList = []

turnOn = True

tableValue = [[0, 0, 0, 0]]

# update function

def stream(i):

# update the main graph(contains actual value and predicted value)

# add the data

global turnOn, highlightList, ax3

x1.append(timestamp[i])

y1.append(value_a[i])

# update the axis

minAxis = max(x1) - datetime.timedelta(days=1)

ax1.set_xlim(minAxis, max(x1))

ax1.set_ylim(min(y1), max(y1))

ax1.figure.canvas.draw()

p1.set_data(x1, y1)

# update the anomaly graph(contains anomaly score)

x2.append(timestamp[i])

y2.append(anomalyScore_a[i])

ax2.set_xlim(minAxis, max(x2))

ax2.set_ylim(min(y2), max(y2))

# update the scatter

if anomalyScore_a[i] >= 0.8:

x1_3.append(timestamp[i])

y1_3.append(value_a[i])

ax1.scatter(x1_3, y1_3, s=50, color="black")

# update the high light

if anomalyScore_a[i] >= 0.8:

highlightList.append(i)

turnOn = True

else:

turnOn = False

if len(highlightList) != 0 and turnOn is False:

ax2.axvspan(timestamp[min(highlightList)] - datetime.timedelta(minutes=10),

timestamp[max(highlightList)] + datetime.timedelta(minutes=10),

color='r',

edgecolor=None,

alpha=0.2)

highlightList = []

turnOn = True

p2.set_data(x2, y2)

# add the table in ax3

# update the anomaly tabel

if anomalyScore_a[i] >= 0.8:

ax3.remove()

ax3 = fig.add_axes([0.8, 0.1, 0.2, 0.8], frameon=False)

ax3.text(0.05, 0.99, "anomaly value table", size=12)

ax3.set_xticks([])

ax3.set_yticks([])

tableValue.append([timestamp[i].strftime("%Y-%m-%d %H:%M:%S"), value_a[i], predict_a[i], anomalyScore_a[i]])

if len(tableValue) >= 40: tableValue.pop(0)

ax3.table(cellText=tableValue, colWidths=[0.35] * 4, colLabels=col_labels, loc=1, cellLoc="center")

return p1, p2

# main animated function

anim = FuncAnimation(fig, stream, init_func=init, frames=len(timestamp), interval=0)

plt.show()

file.close()

ion版本

#! /usr/bin/python

import os

import csv

import datetime

import matplotlib.pyplot as plt

from matplotlib.animation import FuncAnimation

from matplotlib.dates import DateFormatter

import matplotlib.ticker as ticker

class streamDetectionPlot(object):

"""

Anomaly plot output.

"""

# initial the figure parameters.

def __init__(self):

# Turn matplotlib interactive mode on.

plt.ion()

# initial the plot variable.

self.timestamp = []

self.actualValue = []

self.predictValue = []

self.anomalyScore = []

self.tableValue = [[0, 0, 0, 0]]

self.highlightList = []

self.highlightListTurnOn = True

self.anomalyScoreRange = [0, 1]

self.actualValueRange = [0, 1]

self.predictValueRange = [0, 1]

self.timestampRange = [0, 1]

self.anomalyScatterX = []

self.anomalyScatterY = []

# initial the figure.

global fig

fig = plt.figure(figsize=(18, 8), facecolor="white")

fig.subplots_adjust(left=0.06, right=0.70)

self.actualPredictValueGraph = fig.add_subplot(2, 1, 1)

self.anomalyScoreGraph = fig.add_subplot(2, 1, 2)

self.anomalyValueTable = fig.add_axes([0.8, 0.1, 0.2, 0.8], frameon=False)

# define the initial plot method.

def initPlot(self):

# initial two lines of the actualPredcitValueGraph.

self.actualLine, = self.actualPredictValueGraph.plot_date(self.timestamp, self.actualValue, fmt="-",

color="red", label="actual value")

self.predictLine, = self.actualPredictValueGraph.plot_date(self.timestamp, self.predictValue, fmt="-",

color="blue", label="predict value")

self.actualPredictValueGraph.legend(loc="upper right", frameon=False)

self.actualPredictValueGraph.grid(True)

# initial two lines of the anomalyScoreGraph.

self.anomalyScoreLine, = self.anomalyScoreGraph.plot_date(self.timestamp, self.anomalyScore, fmt="-",

color="red", label="anomaly score")

self.anomalyScoreGraph.legend(loc="upper right", frameon=False)

self.baseline = self.anomalyScoreGraph.axhline(0.8, color='black', lw=2)

# set the x/y label of the first two graph.

self.anomalyScoreGraph.set_xlabel("datetime")

self.anomalyScoreGraph.set_ylabel("anomaly score")

self.actualPredictValueGraph.set_ylabel("value")

# configure the anomaly value table.

self.anomalyValueTableColumnsName = ["timestamp", "actual value", "expect value", "anomaly score"]

self.anomalyValueTable.text(0.05, 0.99, "Anomaly Value Table", size=12)

self.anomalyValueTable.set_xticks([])

self.anomalyValueTable.set_yticks([])

# axis format.

self.dateFormat = DateFormatter("%m/%d %H:%M")

self.actualPredictValueGraph.xaxis.set_major_formatter(ticker.FuncFormatter(self.dateFormat))

self.anomalyScoreGraph.xaxis.set_major_formatter(ticker.FuncFormatter(self.dateFormat))

# define the output method.

def anomalyDetectionPlot(self, timestamp, actualValue, predictValue, anomalyScore):

# update the plot value of the graph.

self.timestamp.append(timestamp)

self.actualValue.append(actualValue)

self.predictValue.append(predictValue)

self.anomalyScore.append(anomalyScore)

# update the x/y range.

self.timestampRange = [min(self.timestamp), max(self.timestamp)+datetime.timedelta(minutes=10)]

self.actualValueRange = [min(self.actualValue), max(self.actualValue)+1]

self.predictValueRange = [min(self.predictValue), max(self.predictValue)+1]

# update the x/y axis limits

self.actualPredictValueGraph.set_ylim(

min(self.actualValueRange[0], self.predictValueRange[0]),

max(self.actualValueRange[1], self.predictValueRange[1])

)

self.actualPredictValueGraph.set_xlim(

self.timestampRange[1] - datetime.timedelta(days=1),

self.timestampRange[1]

)

self.anomalyScoreGraph.set_xlim(

self.timestampRange[1]- datetime.timedelta(days=1),

self.timestampRange[1]

)

self.anomalyScoreGraph.set_ylim(

self.anomalyScoreRange[0],

self.anomalyScoreRange[1]

)

# update the two lines of the actualPredictValueGraph.

self.actualLine.set_xdata(self.timestamp)

self.actualLine.set_ydata(self.actualValue)

self.predictLine.set_xdata(self.timestamp)

self.predictLine.set_ydata(self.predictValue)

# update the line of the anomalyScoreGraph.

self.anomalyScoreLine.set_xdata(self.timestamp)

self.anomalyScoreLine.set_ydata(self.anomalyScore)

# update the scatter.

if anomalyScore >= 0.8:

self.anomalyScatterX.append(timestamp)

self.anomalyScatterY.append(actualValue)

self.actualPredictValueGraph.scatter(

self.anomalyScatterX,

self.anomalyScatterY,

s=50,

color="black"

)

# update the highlight of the anomalyScoreGraph.

if anomalyScore >= 0.8:

self.highlightList.append(timestamp)

self.highlightListTurnOn = True

else:

self.highlightListTurnOn = False

if len(self.highlightList) != 0 and self.highlightListTurnOn is False:

self.anomalyScoreGraph.axvspan(

self.highlightList[0] - datetime.timedelta(minutes=10),

self.highlightList[-1] + datetime.timedelta(minutes=10),

color="r",

edgecolor=None,

alpha=0.2

)

self.highlightList = []

self.highlightListTurnOn = True

# update the anomaly value table.

if anomalyScore >= 0.8:

# remove the table and then replot it

self.anomalyValueTable.remove()

self.anomalyValueTable = fig.add_axes([0.8, 0.1, 0.2, 0.8], frameon=False)

self.anomalyValueTableColumnsName = ["timestamp", "actual value", "expect value", "anomaly score"]

self.anomalyValueTable.text(0.05, 0.99, "Anomaly Value Table", size=12)

self.anomalyValueTable.set_xticks([])

self.anomalyValueTable.set_yticks([])

self.tableValue.append([

timestamp.strftime("%Y-%m-%d %H:%M:%S"),

actualValue,

predictValue,

anomalyScore

])

if len(self.tableValue) >= 40: self.tableValue.pop(0)

self.anomalyValueTable.table(cellText=self.tableValue,

colWidths=[0.35] * 4,

colLabels=self.anomalyValueTableColumnsName,

loc=1,

cellLoc="center"

)

# plot pause 0.0001 second and then plot the next one.

plt.pause(0.0001)

plt.draw()

def close(self):

plt.ioff()

plt.show()

下面是ion版本的調(diào)用：

graph = stream_detection_plot.streamDetectionPlot()

graph.initPlot()

for i in xrange(len(timestamp)):

graph.anomalyDetectionPlot(timestamp[i],value_a[i],predict_a[i],anomalyScore_a[i])

graph.close()

具體為實(shí)例化類(lèi)，初始化圖形，傳入數(shù)據(jù)作圖，關(guān)掉。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

總結(jié)

以上是生活随笔為你收集整理的python 可视化监控平台_python可视化篇之流式数据监控的实现的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。