Matplotlib教程 - 学习如何使用Matplotlib和InfluxDB可视化时间序列数据

作者：社区 / 用例，产品，开发者
2022年6月22日

导航到

本文由Armstrong Asenavi撰写。向下滚动查看作者简介。

时间序列是一组按时间顺序排列、时间间隔相等的观测数据点（数据）。一些著名的时间序列数据例子包括股票价格、年降雨记录或每日使用共享单车应用的用户数量。时间序列数据表现出某些模式，例如酒店价格随季节变化的起伏。

InfluxDB是一个领先的高性能时间序列数据库（TSDB），它赋予开发者利用时间序列数据构建实时分析、物联网和云应用程序的能力。

在这篇文章中，您将学习如何利用InfluxDB的功能以及Matplotlib的灵活性和强大功能，来高效地可视化时间序列数据。

InfluxDB是什么？

InfluxDB是一个开源的非关系型TSDB，使用Golang编程语言编写，优化了高可用性和快速读写功能。更快的检索和存储使其在许多应用中变得流行，例如在生产中监控应用程序、存储物联网传感器数据和执行实时分析。InfluxDB提供了一系列访问选项，包括类似SQL的查询。

InfluxDB非常适合存储一旦到达数据库就需要快速处理的数据。因此，它利用网络时间协议（NTP）在系统间同步时间。它还采用了一种简单的索引方法，与其他TSDB相比，提供了明显的速度优势。

InfluxDB适用于涉及大量时间戳数据的各种应用程序。以下是一些企业如何使用InfluxDB进行各种目的的示例

  • 监控生产线信号。德州仪器（TI）使用InfluxDB来监控制造操作并在问题变得昂贵之前检测异常。TI监视了可能成为问题的一千多个不同信号。目标是构建能够识别关键问题并自主行动的应用程序。

  • 提高工业物联网（IIoT）的连接效率。ThingWorx Kepware使用InfluxDB帮助公司将其遗留资产连接到数据库（本地和云）。公司可以轻松且大规模地收集和存储标签数据，使管理者能够实时监控多台设备的事件。

  • 实时监控用户统计。Index Exchange使用InfluxDB来监控用户统计。InfluxDB集群触发器允许实时存储数据，然后分析师可以以低成本实时访问和处理数据。

使用Matplotlib和InfluxDB可视化时间序列数据

现在，让我们更深入地了解如何使用Matplotlib和InfluxDB来可视化时间序列数据。您将了解Matplotlib的基础知识，回顾一些示例图表，然后通过一个教程向您展示如何设置和使用InfluxDB Python客户端来执行CRUD操作。

Matplotlib简介

Matplotlib 是一个开源的数据可视化库，可以用于创建图表、图形和图形。它可以说是 Python 最受欢迎的绘图库，被世界各地的数据科学家和机器学习工程师广泛使用。

在 Matplotlib 中，图形是分层的，通过嵌套 Python 对象来创建树状结构。每个图形都被一个 figure 对象封装，如图所示

Matplotlib figure structure

这个“图形”是可视化的顶层容器。它可以包含多个坐标轴，这些坐标轴基本上是容器内的独立图形。

Matplotlib 使用名为 pyplot 的 API 来简化用户创建可视化的过程——您不需要自己显式配置 figure 和 axes。

您还可以找到控制坐标轴、刻度标记、图例、标题、文本框、网格等 Python 对象，所有这些都可以进行自定义。

Matplotlib 图形的结构如下所示

Matplotlib-Figure-parts

图片来自《数据可视化工作坊》

您可以使用 pip 或 conda 软件包管理器安装 Matplotlib，具体取决于您的喜好

pip install matplotlib

或者

conda install matplotlib

然后按照以下方式安装 Numpy 模块

pip install numpy

使用别名 plt 来引用导入的子模块也是一个便捷的做法

import matplotlib.pyplot as plt

import numpy as np # for creating arrays

import pandas as pd # for manipulating dataframes

使用 Matplotlib，您可以创建各种类型的可视化，如条形图、饼图、雷达图、直方图和散点图。以下是创建一些基本图表类型的几个示例

折线图

plt.plot([1, 2, 3], label='Label 1')

plt.plot([2, 4, 3], label='Label 2')

# Add title

plt.title('Two-way line plot')

plt.legend()

plt.show()

Line Plot in Matplotlib

散点图

# Create a Figure with dpi set to 100

plt.figure(dpi=100)

# Fixing random state for reproducibility

np.random. seed(100)

n = 20

x = np.random.rand(n)

y = np.random.rand(n)

colors = np.random.rand(n)

area = (30 * np.random.rand(n))**2

plt.scatter(x, y, s=area, c=colors, alpha=0.5)

# Add title

plt.title('Scatter plot')

plt.show()

Scatter-plot-in-Matplotlib

条形图

labels = ['A', 'B', 'C', 'D']

x = np.arange(len(labels))

width = 0.4

plt.bar(x - width / 2, [20, 25, 40, 10], width=width)

plt.bar(x + width / 2, [30, 15, 30, 20], width=width)

# Ticks and tick labels must be set manually

plt.xticks(x)

ax = plt.gca()

ax.set_xticklabels(labels)

# Add axes and titles

ax.set_xlabel("x-axis")

ax.set_ylabel("y-axis")

plt.title('Bar chart')

# Show plot

plt.show()

Bar chart in Matplotlib

饼图

# Create figure

plt.figure(figsize=(8, 8), dpi=100)

plt.pie([19.5, 12.2, 4.9, 3.7], explode=(0.1, 0, 0, 0), \

labels=['USA', 'China', 'Japan', 'Germany'], autopct='%.0f%%')

# Add title

plt.title('Top 4 Global Economies by GDP')

# Show plot

plt.show()

Pie-chart-in-Matplotlib

配置 InfluxDB

现在让我们深入了解本教程的细节，本教程将侧重于 InfluxDB 的云版本。由于有各种用于收集、评估和可视化的模块，因此入门非常简单。

首先，在 InfluxDB Cloud 网站上注册免费用户。请确保通过点击发送到您的电子邮件的链接来验证您的账户。然后登录并选择云提供商（AWS、Azure 或 GCP）。请注意，免费版本提供的读写限制有限，最多 10,000 个数据集，保留策略为三十天。

注册后，您应该会出现在仪表板页面

InfluxDB UI

接下来，导航到 桶 部分，并创建一个桶

Creating a bucket in influxDB

导航到 API 令牌 部分，并创建一个 所有访问 API 令牌。这很重要，因为您将向您的桶中写入数据；所有访问令牌将允许您无限制地读取和写入桶中的数据。

然后您将想要创建所需的文件夹，设置 Python 虚拟环境和安装所需的包

# Create a directory

mkdir influxDB-Tutorial

# Navigate into the folder

cd influxDB-Tutorial

接下来，使用 venv 库创建虚拟环境。请注意，当运行 Python v3.6+ 时，推荐使用 venv。有关更多信息，请参阅 此文档。

# To create a virtual environment

pip install venv

# Activate virtual environment

source venv/Scripts/activate

# Install influxDB client

pip install influxdb-client

# Create a file known as named __init__.py

touch __init__.py

# Create a file known a file .env to store credentials

touch .env

# Create .gitignore file to prevent .env going public

touch .gitignore

# Install python-dotenv to allow access to the .env file

pip install python-dotenv

# Create folders for storing data and img

mkdir data

mkdir img

# Install yfinance to Collect data

pip install yfinance

# Install matplotlib

pip install matplotlib

将凭据（令牌、桶和组织）存储在 .env 文件中。请记住在 .gitignore 文件中包含 .env 文件。

INFLUX_TOKEN = " All access token from InfluxDB"

ORG = "email you registered with on InfluxDB"

BUCKET = "name of your bucket"

在 __init__.py 文件中，开始导入 dotenv 和 os 模块以从 .env 文件中读取凭据。

然后返回到 InfluxDB Cloud UI，转到 源 部分。在客户端库中，选择“Python”。复制 初始化客户端 下的代码，并将其粘贴到 __init__.py 文件中

Initialize the Client

# Import modules

from dotenv import load_dotenv

import os

from influxdb_client import InfluxDBClient, Point, WriteOptions

from influxdb_client.client.write_api import SYNCHRONOUS

load_dotenv()

token = os.getenv('INFLUX_TOKEN')

org = os.getenv('ORG')

bucket = os.getenv('BUCKET')

client = InfluxDBClient(url="https://us-east-1-1.aws.cloud2.influxdata.com", token=token, org=org)

接下来，运行您的 Python 文件

$ python __init__.py

如果没有错误信息，您已成功连接到 InfluxDB Cloud。

使用 Matplotlib 与 InfluxDB

本教程使用从雅虎财经下载的调整后收盘价，对四家科技公司（苹果（AAPL）、亚马逊（AMZN）、谷歌（GOOG）和微软（MSFT））的股票进行可视化分析，时间范围从2020年1月1日至2022年3月30日。目的是可视化COVID-19疫情期间股票的价格趋势。

下载和保存数据

首先，使用yfinance库下载数据

import yfinance as yf

tickers = ['AAPL', 'GOOG', 'AMZN', 'MSFT']

#Download only adjusted closing price for the tickers

data = yf.download(tickers, start="2020-01-01", end="2022-03-30")['Adj Close']

data.head()

导入的数据将如下所示

Output-table-stock-data

然后使用以下代码将数据保存为CSV文件

# Save the data into the data folder as a csv file.

data.to_csv('data/stocks.csv')

准备写入数据

接下来，使用以下代码加载所需的库并链接到InfluxDB Python客户端

# Import required packages and access influxdb client

from collections import OrderedDict

from csv import DictReader

import rx

from rx import operators as ops

from dotenv import load_dotenv

import os

from influxdb_client import InfluxDBClient, Point, WriteOptions

from influxdb_client.client.write_api import SYNCHRONOUS

load_dotenv()

token = os.getenv('INFLUX_TOKEN')

org = os.getenv('ORG')

bucket = os.getenv('BUCKET')

client = InfluxDBClient(url="https://us-east-1-1.aws.cloud2.influxdata.com", token=token, org=org)

创建一个写入数据的函数

然后使用以下代码将CSV数据准备成点结构，并将其写入名为finance-bucket的InfluxDB桶中

def parse_row(row: OrderedDict):

"""

This function parses rows into Point with structure:

the csv file has the following columns:

Date,AAPL,AMZN,GOOG,MSFT

...

:param row: row of csv file

:return: Parsed csv row to Point

"""

return Point("financial-analysis")\

.tag("type", "stock-daily") \

.field("aapl", float(row['AAPL'])) \

.field("amzn", float(row['AMZN'])) \

.field("goog", float(row['GOOG'])) \

.field("msft", float(row['MSFT'])) \

.time(row['Date'])

"""

Convert the stocks.csv into data points

"""

data = rx \

.from_iterable(DictReader(open('./data/stocks.csv', 'r'))) \

.pipe(ops.map(lambda row: parse_row(row)))

"""

Create a client that writes data.

"""

write_api = client.write_api(write_options=WriteOptions(batch_size=5_000, flush_interval=1_000))

"""

Write data into InfluxDB

"""

write_api.write(bucket=bucket, record=data)

write_api.close()

使用Flux从InfluxDB Cloud查询数据

一旦数据在桶中，您可以使用以下代码使用Flux语言查询它，并返回一个dataframe（df）

query_with_multiple_fields ='''

from(bucket: "finance-bucket")

|> range(start:0, stop: now())

|> filter(fn: (r) => r._measurement == "financial-analysis")

'''

# Output the data as data frame named df

df = client.query_api().query_data_frame(org=org, query=query_with_multiple_fields)

df.head()

如您所见，dataframe中的数据以扁平结构呈现，每行一个字段

Flux Query 1 output

编写Flux查询以将时间序列转换为dataframe

接下来，使用Flux查询并旋转数据，使其以所有字段作为列呈现的dataframe

query_with_pivot ='''

from(bucket: "finance-bucket")

|> range(start:0, stop: now())

|> filter(fn: (r) => r._measurement == "financial-analysis")

|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")

|> keep(columns:["_time", "aapl", "amzn", "goog", "msft"])

'''

# put the data in a dataframe

df = client.query_api().query_data_frame(org=org, query=query_with_pivot)

df.head()

输出显示数据现在已按照预期在dataframe中。然而，如以下截图所示，InfluxDB添加了“result”和“table”列，您将在下一节中删除这些列。

Flux-Query-2-output

删除无关列

使用drop 方法删除不相关的“result”和“table”列

df["_time"] = pd.to_datetime(df["_time"].astype(str))

df = df.drop(columns=["result", "table"])

df = df.set_index("_time")

df.head()

最终的dataframe呈现如下

Preprocessed-dataframe-for-analysis

规范化dataframe以创建专业图表

dataframe显示，AMZN和GOOG的股价在数千美元，而AAPL和MSFT的股价在数十和数百美元。以这种方式绘制数据将导致顶部有两个图表（AMZN和GOOG）和底部有两个图表（MSFT和AAPL），这看起来不够专业。为了解决这个问题，您可以针对第一天规范化股票数据，强制第一天股价从1美元开始。

使用以下代码实现

def normalize_df(df):

"""normalizes stock data w.r.t price in day 1,

force first day price to start at $1"""

return df/df.iloc[0,:]

data = normalize_df(df)

使用Matplotlib可视化数据

最后，使用以下代码创建股票价格的时间序列图

# Create the plot

ax = data.plot(figsize=(15, 10))

ax.set_xlabel("Date")

ax.set_ylabel("Adjusted Price")

plt.title('Adjusted Closing Prices')

plt.legend(loc='upper left', fontsize=12)

plt.tight_layout()

plt.style.use('bmh')

plt.grid(True)

plt.show()

运行代码后，输出将如下所示

Time Series Plot

数据分析

如您在最终图中所见，2020年3月至4月，由于COVID-19疫情宣布封锁措施，股价出现暂时性下跌。从那时起，尽管COVID-19的影响，科技股表现相对较好。有许多选项可以进行更深入的时间序列数据分析，以及如何通过时间序列预测方法进行预测。

结论

在本教程中，您学习了Matplotlib（Python中流行的绘图库）的时间序列可视化的基础知识。您还学习了如何设置InfluxDB Cloud，将数据写入桶，以及如何使用简单的Python脚本将数据读入Python进行分析，通过InfluxDB执行读写操作。

正如您所看到的，结合InfluxDB和Matplotlib的功能，可以创建高效的时序数据处理可视化，拥有强大的功能和灵活性。您可以在这个GitHub仓库中访问本教程的完整源代码。

附加资源

如果您对其他使用InfluxDB进行数据可视化的教程感兴趣，请查看以下链接

  • Recharts教程 - 学习如何使用Recharts图表库和InfluxDB可视化物联网数据

  • Highcharts InfluxDB教程 - 学习如何使用Highcharts JavaScript图表库和InfluxDB可视化海洋潮汐数据

  • React Native Victory教程 - 本教程展示了如何使用React Native版本的Victory图表库，允许您利用您的Web开发经验创建原生移动应用

  • Nivo教程- 本文介绍了Nivo图表库的基本原理，以及如何从InfluxDB中提取数据创建各种类型的图表

  • Plotly.js教程- 使用PlotlyJS图表库显示存储在InfluxDB中的数据

关于作者

Armstrong Asenavi是一位经验丰富的机器学习工程师和数据分析师，对洞察力、商业智能（BI）、仪表盘、指标、分析、建模和可视化有浓厚的兴趣。他具备提升数据工程项目价值各个阶段的相关经验，熟练掌握Python、R、JavaScript、SQL和DAX等主要编程语言。他在研究、项目管理和技术报告撰写方面有着坚实的背景。他是一位分析型、批判性思维者和数据驱动型分析师。Armstrong沟通清晰，能够将复杂信息转化为可行解决方案。他与团队成员合作良好，在团队环境中茁壮成长。