使用PyArrow库分析数据

使用 PyArrow 来读取和分析来自 InfluxDB Cloud Dedicated 的查询结果。 PyArrow 库提供高效的计算、聚合、序列化和 Arrow 格式数据的转换。

Apache Arrow 是一个用于内存分析的开发平台。它包含一套技术，使大数据系统能够快速存储、处理和移动数据。

Arrow Python 绑定（也称为“PyArrow”）与 NumPy、pandas 和内置 Python 对象具有一级集成。它们基于 Arrow 的 C++ 实现。

PyArrow documentation

• 安装先决条件
• 使用 PyArrow 读取查询结果
• 使用PyArrow分析数据
  • 分组和聚合数据

安装前提条件

本指南中的示例假设使用 Python 虚拟环境和 InfluxDB 3 influxdb3-python Python 客户端库。有关更多信息，请参见如何 开始使用 Python 查询 InfluxDB。

安装 influxdb3-python 也会安装 pyarrow 库，该库为 Apache Arrow 提供 Python 绑定。

使用 PyArrow 读取查询结果

以下示例展示了如何使用 influxdb3-python 和 pyarrow 查询 InfluxDB，并将 Arrow 数据视为 PyArrow Table。

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

   # pyarrow-example.py
   
   from influxdb_client_3 import InfluxDBClient3
   import pandas
   
   def querySQL():
     
     # Instantiate an InfluxDB client configured for a database
     client = InfluxDBClient3(
       "https://cluster-id.a.influxdb.io",
       database="
   DATABASE_NAME
   ",
       token="
   DATABASE_TOKEN
   ")
   
     # Execute the query to retrieve all record batches in the stream formatted as a PyArrow Table.
     table = client.query(
       '''SELECT *
         FROM home
         WHERE time >= now() - INTERVAL '90 days'
         ORDER BY time'''
     )
   
     client.close()
   
   print(querySQL())
   

2. 替换以下配置值：

   • DATABASE_TOKEN: 一个InfluxDB 数据库令牌 具有您希望查询的数据库的读取权限。
   • DATABASE_NAME: 要查询的 InfluxDB 数据库 的名称。

3. 在你的终端中，使用Python解释器运行文件：

   python pyarrow-example.py
   

该 InfluxDBClient3.query() 方法发送查询请求，然后返回一个 pyarrow.Table，其中包含来自响应流的所有 Arrow 记录批次。

接下来， 使用 PyArrow 分析数据。

使用 PyArrow 分析数据

分组和汇总数据

使用 pyarrow.Table，您可以将列中的值作为 键 进行分组。

以下示例展示了如何查询InfluxDB，然后使用PyArrow对表数据进行分组并计算每个组的聚合值：

# pyarrow-example.py

from influxdb_client_3 import InfluxDBClient3
import pandas

def querySQL():
  
  # Instantiate an InfluxDB client configured for a database
  client = InfluxDBClient3(
    "https://cluster-id.a.influxdb.io",
    database="
DATABASE_NAME
",
    token="
DATABASE_TOKEN
")

  # Execute the query to retrieve data 
  # formatted as a PyArrow Table
  table = client.query(
    '''SELECT *
      FROM home
      WHERE time >= now() - INTERVAL '90 days'
      ORDER BY time'''
  )

  client.close()

  return table

table = querySQL()

# Use PyArrow to aggregate data
print(table.group_by('room').aggregate([('temp', 'mean')]))

替换以下内容：

• DATABASE_TOKEN: 一个具有查询权限的 InfluxDB 数据库令牌。
• DATABASE_NAME: 要查询的InfluxDB 数据库的名称。

pyarrow.Table
temp_mean: double
room: string
----
temp_mean: [[22.581987577639747,22.10807453416151]]
room: [["Kitchen","Living Room"]]

有关更多详细信息和示例，请参阅PyArrow 文档和Apache Arrow Python 食谱。