使用 Python 查询数据

使用 InfluxDB influxdb_client_3 Python 客户端库模块和 SQL 或 InfluxQL 查询存储在 InfluxDB 中的数据。通过 Flight+gRPC 协议执行查询并检索数据，然后使用常用的 Python 工具处理数据。

• 开始使用 Python 查询 InfluxDB
• 创建 Python 虚拟环境
  • 安装 Python
  • 创建项目虚拟环境
  • 安装 Anaconda
• 查询 InfluxDB
  • 安装 influxdb3-python 库
  • 创建 InfluxDB 客户端
  • 执行查询

开始使用 Python 查询 InfluxDB

本指南假定具备以下先决条件

• 一个带有要查询数据的 InfluxDB Clustered 数据库
• 一个具有对数据库读取权限的数据库令牌

要了解如何设置 InfluxDB 和写入数据，请参阅“入门教程”中的设置说明。

创建 Python 虚拟环境

本指南遵循使用 Python 虚拟环境的推荐做法。如果您不想使用虚拟环境并且已安装 Python，请继续查询 InfluxDB。 Python 虚拟环境使您的项目的 Python 解释器和依赖项保持独立，并与其他项目隔离。

要安装 Python 并创建虚拟环境，请选择以下选项之一

• Python venv：venv 模块在 Python 3.5 及更高版本中是标准模块。

• Anaconda® Distribution：一个 Python/R 数据科学发行版，提供 Python 和 conda 包和环境管理器。

  venv Anaconda

  安装 Python

  1. 按照 Python 安装说明 为您的系统安装最新版本的 Python 编程语言。

  2. 检查您是否可以运行 python 和 pip 命令。 pip 是大多数 Python 发行版中包含的包管理器。

     在您的终端中，输入以下命令

     python --version
     

     pip --version
     

     根据您的系统，您可能需要使用特定于版本的命令，例如。

     python3 --version
     

     pip3 --version
     

     如果 pip 和 pip<PYTHON_VERSION> 都不起作用，请按照 Pypa.io Pip 安装方法之一为您的系统安装。

  创建项目虚拟环境

  1. 为您的 Python 项目创建一个目录，并切换到新目录，例如

     mkdir ./PROJECT_DIRECTORY && cd $_
     

  2. 使用 Python venv 模块创建一个虚拟环境，例如

     python -m venv envs/virtualenv-1
     

     venv 在您的项目中创建新的虚拟环境目录。

  3. 要在您的终端中激活新的虚拟环境，请运行 source 命令并传递虚拟环境 activate 脚本的路径

     source envs/VIRTUAL_ENVIRONMENT_NAME/bin/activate
     

     例如

     source envs/virtualenv-1/bin/activate
     

  安装 Anaconda

  1. 按照 Anaconda 安装说明 为您的系统安装。

  2. 检查您是否可以运行 conda 命令

     conda
     

  3. 使用 conda 创建虚拟环境，例如

     conda create --prefix envs/virtualenv-1 
     

     conda 在名为 ./envs/virtualenv-1 的目录中创建一个虚拟环境。

  4. 要激活新的虚拟环境，请使用 conda activate 命令并传递虚拟环境的目录路径

     conda activate envs/VIRTUAL_ENVIRONMENT_NAME
     

     例如

     conda activate ./envs/virtualenv-1
     

当虚拟环境被激活时，名称会显示在终端命令行开头，例如

(virtualenv-1) $ PROJECT_DIRECTORY

查询 InfluxDB

1. 安装 influxdb3-python 库
2. 创建 InfluxDB 客户端
3. 执行查询

安装 influxdb3-python 库

influxdb3-python 包提供了 influxdb_client_3 模块，用于将 InfluxDB Clustered 与您的 Python 代码集成。该模块支持将数据写入 InfluxDB，以及使用 SQL 或 InfluxQL 查询数据。

安装以下依赖项

* 已在写入数据部分中安装

• influxdb3-python *：提供 influxdb_client_3 模块，并安装 pyarrow 包，用于处理从查询返回的 Arrow 数据。
• pandas：提供 pandas 模块，用于分析和操作数据。
• tabulate：提供 tabulate 函数，用于格式化表格数据。

在您的终端中输入以下命令

pip install influxdb3-python pandas tabulate

安装 influxdb3-python 和 pyarrow 后，您就可以查询和分析存储在 InfluxDB 数据库中的数据了。

创建 InfluxDB 客户端

以下示例演示了如何将 Python 与 influxdb_client_3 模块结合使用，以实例化配置为 InfluxDB Clustered 数据库的客户端。

在您的编辑器中，复制以下示例代码并粘贴到一个新文件中，例如 query-example.py

# query-example.py

from influxdb_client_3 import InfluxDBClient3

# Instantiate an InfluxDBClient3 client configured for your database
client = InfluxDBClient3(
    host='cluster-host.com',
    token='
DATABASE_TOKEN
',
    database='
DATABASE_NAME
'
)

from influxdb_client_3 import InfluxDBClient3, flight_client_options
import certifi

fh = open(certifi.where(), "r")
cert = fh.read()
fh.close()

client = InfluxDBClient3(
host="cluster-host.com",
token='
DATABASE_TOKEN
',
database='
DATABASE_NAME
',
flight_client_options=flight_client_options(
tls_root_certs=cert))
...

替换以下配置值

• database：要查询的 InfluxDB Clustered 数据库的名称
• token：具有对指定数据库读取权限的数据库令牌。将其存储在密钥库或环境变量中，以避免暴露原始令牌字符串。

执行查询

要执行查询，请调用以下客户端方法

query(query,language) 方法

并指定以下参数

• query：字符串。要执行的 SQL 或 InfluxQL 查询。
• language：字符串（"sql" 或 "influxql"）。query 语言。

示例

以下示例演示了如何使用 SQL 或 InfluxQL 选择测量中的所有字段，然后将结果格式化为 Markdown 表格输出。

# query-example.py

from influxdb_client_3 import InfluxDBClient3

client = InfluxDBClient3(
    host='cluster-host.com',
    token='
DATABASE_TOKEN
',
    database='
DATABASE_NAME
'
)

# Execute the query and return an Arrow table
table = client.query(
    query="SELECT * FROM home",
    language="sql"
)

print("\n#### View Schema information\n")
print(table.schema)
print(table.schema.names)
print(table.schema.types)
print(table.field('room').type)
print(table.schema.field('time').metadata)

print("\n#### View column types (timestamp, tag, and field) and data types\n")
print(table.schema.field('time').metadata[b'iox::column::type'])
print(table.schema.field('room').metadata[b'iox::column::type'])
print(table.schema.field('temp').metadata[b'iox::column::type'])

print("\n#### Use PyArrow to read the specified columns\n")
print(table.column('temp'))
print(table.select(['room', 'temp']))
print(table.select(['time', 'room', 'temp']))

print("\n#### Use PyArrow compute functions to aggregate data\n")
print(table.group_by('hum').aggregate([]))
print(table.group_by('room').aggregate([('temp', 'mean')]))

# query-example.py

from influxdb_client_3 import InfluxDBClient3

client = InfluxDBClient3(
    host='cluster-host.com',
    token='
DATABASE_TOKEN
',
    database='
DATABASE_NAME
'
)

# Execute the query and return an Arrow table
table = client.query(
    query="SELECT * FROM home",
    language="influxql"
)

print("\n#### View Schema information\n")
print(table.schema)
print(table.schema.names)
print(table.schema.types)
print(table.field('room').type)
print(table.schema.field('time').metadata)

print("\n#### View column types (timestamp, tag, and field) and data types\n")
print(table.schema.field('time').metadata[b'iox::column::type'])
print(table.schema.field('room').metadata[b'iox::column::type'])
print(table.schema.field('temp').metadata[b'iox::column::type'])

print("\n#### Use PyArrow to read the specified columns\n")
print(table.column('temp'))
print(table.select(['room', 'temp']))
print(table.select(['time', 'room', 'temp']))

print("\n#### Use PyArrow compute functions to aggregate data\n")
print(table.group_by('hum').aggregate([]))
print(table.group_by('room').aggregate([('temp', 'mean')]))

替换以下配置值

• database：要查询的 InfluxDB Clustered 数据库的名称
• token：具有对指定数据库读取权限的数据库令牌。将其存储在密钥库或环境变量中，以避免暴露原始令牌字符串。

接下来，了解如何使用 Python 工具处理时间序列数据

• 使用 PyArrow
• 使用 pandas