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 对象封装了每个绘图，如图所示
这个“figure”是可视化的顶层容器。它可以有多个轴，这些轴基本上是容器内的单独绘图。
Matplotlib 使用一个名为 pyplot 的 API，使用户更容易创建可视化效果——您不必显式配置 figure 和 axes 本身。
您还可以找到控制轴、刻度线、图例、标题、文本框、网格和许多其他内容的 Python 对象，所有这些都可以自定义。
Matplotlib 绘图的解剖结构如下
图片由 The Data Visualization Workshop 提供
可以使用 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()
散点图
# 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()
条形图
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()
饼图
# 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()
设置 InfluxDB
现在让我们深入了解本教程的细节，本教程将重点介绍云版本的 InfluxDB。由于有各种用于收集、评估和可视化数据的模块，因此入门非常容易。
首先，在 InfluxDB Cloud 网站上注册为免费用户。务必单击发送到您电子邮件的链接来验证您的帐户。然后登录并选择云提供商（AWS、Azure 或 GCP）。请注意，免费版本提供有限的读取和写入，最多 10,000 个数据集，以及 30 天的保留策略。
注册后，您应该会看到仪表板页面
接下来，导航到 Buckets 部分并创建一个存储桶
导航到 API Token 部分并创建一个 All Access API Token。这很重要，因为您将数据写入存储桶； All Access Token 将允许您无限制地读取和写入数据到存储桶中。
然后，您需要创建所需的文件夹，设置 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 文件中。请记住将 .env 文件包含在 .gitignore 文件中。
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，转到 Sources 部分。在客户端库中，选择“Python”。复制 Initialize the Client 下的代码并将其粘贴到 __init__.py 文件中
# 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 结合使用
本教程使用来自 Yahoo Finance 的四家科技公司的调整后收盘价：Apple (AAPL)、Amazon (AMZN)、Google (GOOG) 和 Microsoft (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()导入的数据将如下所示
然后使用此代码将数据另存为 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 语言查询它，并使用以下代码返回数据帧 (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()如您所见，数据帧中的数据以平面结构呈现，每一行中都有一个字段
编写 Flux 查询以将时间序列透视为数据帧
接下来，使用 Flux 查询并将数据 透视 到数据帧中，其中所有字段都呈现为列
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()输出显示数据现在位于预定的数据帧中。但是，正如您在下面的屏幕截图中看到的，InfluxDB 添加了“result”和“table”列，您将在下一节中删除这些列。
删除不相关的列
使用 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()最终数据帧呈现如下
规范化数据帧以创建专业的图表
数据帧显示 AMZN 和 GOOG 股票价格以千为单位，而 AAPL 和 MSFT 的股票价格以十和百为单位。按原样绘制数据将在顶部显示两个图表（AMZN 和 GOOG），在底部显示两个图表（MSFT 和 AAPL），这看起来不专业。为了解决这个问题，您可以将股票数据相对于第 1 天进行规范化，强制第一天的价格从 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()运行代码后，输出将如下所示
分析数据
正如您在最终图中看到的，在 2020 年 3 月至 4 月期间，当各国政府宣布因 COVID-19 采取封锁措施时，价格出现暂时下跌。此后，尽管受到 COVID-19 的影响，科技股的表现仍然相对良好。有许多选项可用于更深入的 时间序列数据分析，以及如何通过 时间序列预测 方法进行预测。
结论
在本教程中，您学习了 Matplotlib 中时间序列可视化的基础知识，Matplotlib 是 Python 的流行绘图库。您还学习了如何设置 InfluxDB Cloud、将数据写入存储桶以及将数据读入 Python 进行分析，使用简单的 Python 脚本在 InfluxDB 中执行读写操作。
正如您所见，结合 InfluxDB 和 Matplotlib 的功能来高效地可视化时间序列数据具有强大的功能和灵活性。您可以在 此 GitHub 存储库 中访问本教程的完整源代码。
其他资源
如果您对其他一些教程感兴趣，这些教程展示了如何使用 InfluxDB 可视化数据，请查看以下链接
-
Recharts 教程 - 了解如何使用 Recharts 图表库和 InfluxDB 可视化 IoT 数据
-
Highcharts InfluxDB 教程 - 了解如何将 Highcharts JavaScript 图表库与 InfluxDB 结合使用来可视化海洋潮汐数据。
-
React Native Victory 教程 - 本教程展示了如何使用 Victory 图表库的 React Native 版本，该版本允许您使用您的 Web 开发经验来创建原生移动应用程序。
-
Nivo 教程- 在本文中，您将了解 Nivo 图表库的基础知识，以及如何创建各种类型的图表，同时从 InfluxDB 中提取数据。
-
Plotly.js 教程- 使用 PlotlyJS 图表库显示存储在 InfluxDB 中的数据。
关于作者
Armstrong Asenavi 是一位经验丰富的机器学习工程师和数据分析师，热衷于洞察力、商业智能 (BI)、仪表板、指标、分析、建模和可视化。他在价值提升型数据工程项目的各个阶段都拥有必要的经验，并且精通主要的编程语言，包括 Python、R、JavaScript、SQL 和 DAX。他在研究、项目管理和技术报告撰写方面拥有强大的背景。他是一位分析型、批判性思维者和数据驱动型分析师。 Armstrong 沟通清晰，能够将复杂的信息转化为可行的解决方案。他与团队成员合作良好，并在团队环境中茁壮成长。
