开始查询数据

使用 SQL 和 influx3 CLI 查询 InfluxDB 3。

以下步骤包括设置 Python 虚拟环境，这已在 入门数据写入 中介绍过。如果您的项目的虚拟环境已在运行，请跳至步骤 3。

1. 为您的项目创建一个目录并切换到该目录

   mkdir influx3-query-example && cd $_
   

2. 要创建并激活 Python 虚拟环境，请运行以下命令

   python -m venv envs/virtual-env && . envs/virtual-env/bin/activate
   

3. 安装 CLI 包（已在 数据写入部分 中安装）。

   pip install influxdb3-python-cli
   

   安装 influxdb3-python-cli 还会安装 pyarrow 库，用于处理从查询返回的 Arrow 数据。

4. 创建 config.json 配置文件。

   influx3 config \
     --name="config-clustered" \
     --database="get-started" \
     --host="cluster-host.com" \
     --token="DATABASE_TOKEN" \
     --org="ORG_ID"
   

   替换以下内容

   • DATABASE_TOKEN：具有对 get-started 数据库读取权限的 数据库令牌
   • ORG_ID：任何非空字符串（InfluxDB 忽略此参数，但客户端需要它）

5. 输入 influx3 sql 命令和您的 SQL 查询语句。

   influx3 sql "SELECT *
               FROM home
               WHERE time >= '2022-01-01T08:00:00Z'
               AND time <= '2022-01-01T20:00:00Z'"
   

influx3 在您的终端中显示查询结果。

使用 influxdb_client_3 客户端库模块将 InfluxDB Clustered 与您的 Python 代码集成。客户端库支持将数据写入 InfluxDB 并使用 SQL 或 InfluxQL 查询数据。

以下步骤包括设置 Python 虚拟环境，这已在 入门数据写入 中介绍过。如果您的项目的虚拟环境已在运行，请跳至步骤 3。

1. 在您在 数据写入部分 中创建的 influxdb_py_client 模块目录中打开一个终端

   1. 要创建并激活您的 Python 虚拟环境，请在您的终端中输入以下命令

      python -m venv envs/virtual-env && . ./envs/virtual-env/bin/activate
      

   2. 安装以下依赖项

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

      • influxdb3-python*：提供 InfluxDB influxdb_client_3 Python 客户端库模块，并安装 pyarrow 包，用于处理从查询返回的 Arrow 数据。
      • pandas：提供 pandas 函数、模块和数据结构，用于分析和操作数据。
      • tabulate：提供 tabulate 函数，用于格式化表格数据。pandas 需要此模块才能将数据格式化为 Markdown。

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

      pip install influxdb3-python pandas tabulate
      

   3. 在您的终端或编辑器中，为您的代码创建一个新文件 - 例如：query.py。

2. 在 query.py 中，输入以下示例代码

   from influxdb_client_3 import InfluxDBClient3
   
   client = InfluxDBClient3(
       host=f"cluster-host.com",
       token=f"DATABASE_TOKEN",
       database=f"get-started",
   )
   
   sql = '''
     SELECT
       *
     FROM
       home
     WHERE
       time >= '2022-01-01T08:00:00Z'
       AND time <= '2022-01-01T20:00:00Z'
   '''
   
   table = client.query(query=sql)
   assert table['room'], "Expect table to have room column."
   print(table.to_pandas().to_markdown())
   

示例代码执行以下操作

1. 从 influxdb_client_3 模块导入 InfluxDBClient3 构造函数。

2. 使用凭据调用 InfluxDBClient3() 构造函数方法，以实例化具有以下凭据的 InfluxDB client

   • host：InfluxDB 集群 URL（不带 https:// 协议或尾部斜杠）
   • token：具有对指定数据库读取权限的 数据库令牌。将其存储在密钥存储或环境变量中，以避免暴露原始令牌字符串。
   • database：要查询的 InfluxDB Clustered 数据库的名称

3. 定义要执行的 SQL 查询，并将其分配给 query 变量。

4. 使用 SQL 查询调用 client.query() 方法。query() 向 InfluxDB 发送 Flight 请求，查询数据库，从端点检索结果数据，然后返回分配给 table 变量的 pyarrow.Table。

5. 调用 to_pandas() 方法，将 Arrow 表转换为 pandas.DataFrame。

6. 调用 pandas.DataFrame.to_markdown() 方法，将 DataFrame 转换为 Markdown 表格。

7. 调用 print() 方法，将 Markdown 表格打印到 stdout。

8. 输入以下命令以运行程序并查询您的 InfluxDB 集群

   python query.py
   

1. 在您在 数据写入部分 中创建的 influxdb_go_client 目录中，创建一个名为 query.go 的新文件。

2. 在 query.go 中，输入以下示例代码

   package main
   
   import (
     "context"
     "fmt"
     "io"
     "os"
     "time"
     "text/tabwriter"
   
     "github.com/InfluxCommunity/influxdb3-go/v2/influxdb3"
   )
   
   func Query() error {
   
     // INFLUX_TOKEN is an environment variable you created
     // for your database read token.
     token := os.Getenv("INFLUX_TOKEN")
   
     // Instantiate the client.
     client, err := influxdb3.New(influxdb3.ClientConfig{
       Host:     "https://cluster-host.com",
       Token:    token,
       Database: "get-started",
     })
   
     // Close the client when the function returns.
     defer func(client *influxdb3.Client) {
       err := client.Close()
       if err != nil {
         panic(err)
       }
     }(client)
   
     // Define the query.
     query := `SELECT *
       FROM home
       WHERE time >= '2022-01-01T08:00:00Z'
       AND time <= '2022-01-01T20:00:00Z'`
   
     // Execute the query.
     iterator, err := client.Query(context.Background(), query)
   
     if err != nil {
       panic(err)
     }
   
     w := tabwriter.NewWriter(io.Discard, 4, 4, 1, ' ', 0)
     w.Init(os.Stdout, 0, 8, 0, '\t', 0)
     fmt.Fprintln(w, "time\troom\ttemp\thum\tco")
   
     // Iterate over rows and prints column values in table format.
     for iterator.Next() {
       row := iterator.Value()
       // Use Go time package to format unix timestamp
       // as a time with timezone layout (RFC3339).
       time := (row["time"].(time.Time)).
         Format(time.RFC3339)
       fmt.Fprintf(w, "%s\t%s\t%d\t%.1f\t%.1f\n",
         time, row["room"], row["co"], row["hum"], row["temp"])
     }
   
     w.Flush()
     return nil
   }
   

   示例代码执行以下操作

   1. 导入以下包

      • context
      • fmt
      • io
      • os
      • text/tabwriter
      • github.com/InfluxCommunity/influxdb3-go/v2/influxdb3

   2. 定义一个 Query() 函数，该函数执行以下操作

      1. 使用以下 InfluxDB 凭据参数实例化 influx.Client

         • Host：您的 InfluxDB 集群 URL
         • Database：您的 InfluxDB Clustered 数据库的名称
         • Token：具有对指定数据库读取权限的 数据库令牌。将其存储在密钥存储或环境变量中，以避免暴露原始令牌字符串。

      2. 定义一个延迟函数，以便在执行后关闭客户端。

      3. 为 SQL 查询定义一个字符串变量。

      4. 调用 influxdb3.Client.Query(sql string) 方法，并传递 SQL 字符串来查询 InfluxDB。Query(sql string) 方法返回响应流中数据的 iterator。

      5. 迭代行，将时间戳格式化为 RFC3339 时间戳，并将数据以表格格式打印到 stdout。

3. 在您的编辑器中，打开您在 数据写入部分 中创建的 main.go 文件，并插入代码以调用 Query() 函数 - 例如

   package main
   
   func main() {	
     WriteLineProtocol()
     Query()
   }
   

4. 在您的终端中，输入以下命令以安装必要的包，构建模块并运行程序

   go mod tidy && go run influxdb_go_client
   

   该程序执行 main() 函数，该函数写入数据并将查询结果打印到控制台。

本教程假设您已安装 Node.js 和 npm，并按照 数据写入部分 中的描述创建了 influxdb_js_client npm 项目。

1. 在您的终端或编辑器中，更改为您在 数据写入部分 中创建的 influxdb_js_client 目录。

2. 如果您尚未安装，请安装 @influxdata/influxdb3-client JavaScript 客户端库作为您项目的依赖项

   npm install --save @influxdata/influxdb3-client
   

3. 创建一个名为 query.mjs 的文件。.mjs 扩展名告诉 Node.js 解释器您正在使用 ES6 模块语法。

4. 在 query.mjs 内部，输入以下示例代码

   // query.mjs
   import {InfluxDBClient} from '@influxdata/influxdb3-client'
   import {tableFromArrays} from 'apache-arrow';
   
   /**
   * Set InfluxDB credentials.
   */
   const host = "https://cluster-host.com";
   const database = 'get-started';
   /**
   * INFLUX_TOKEN is an environment variable you assigned to your
   * database READ token value.
   */
   const token = process.env.INFLUX_TOKEN;
   
   /**
   * Query InfluxDB with SQL using the JavaScript client library.
   */
   export async function querySQL() {
     /**
     * Instantiate an InfluxDBClient
     */
     const client = new InfluxDBClient({host, token})
     const sql = `
     SELECT *
     FROM home
     WHERE time >= '2022-01-01T08:00:00Z'
       AND time <= '2022-01-01T20:00:00Z'
     `
   
     const data = {time: [], room: [], co: [], hum: [], temp: []};
     const result = client.query(query, database);
   
     for await (const row of result) {
       data.time.push(new Date(row._time))
       data.room.push(row.room)
       data.co.push(row.co);
       data.hum.push(row.hum);
       data.temp.push(row.temp);
     }
   
     console.table([...tableFromArrays(data)])
   
     client.close()
   }
   

   示例代码执行以下操作

   1. 导入以下内容

      • InfluxDBClient 类
      • tableFromArrays 函数

   2. 调用 new InfluxDBClient() 并传递 ClientOptions 对象以实例化配置有 InfluxDB 凭据的客户端。

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

   3. 为 SQL 查询定义一个字符串变量 (sql)。

   4. 定义一个对象 (data)，其列名作为键，数组值用于存储行数据。

   5. 使用以下参数调用 InfluxDBClient.query() 方法

      • sql：要执行的查询
      • database：要查询的 InfluxDB Clustered 数据库的名称

      query() 返回行向量流。

   6. 迭代行并将列数据添加到 data 中的数组。

   7. 将 data 传递给 Arrow tableFromArrays() 函数以将数组格式化为表格，然后将结果传递给 console.table() 方法以在终端中输出突出显示的表格。

5. 在 index.mjs（在 数据写入部分 中创建）内部，输入以下示例代码以导入模块并调用函数

   // index.mjs
   import { writeLineProtocol } from "./write.mjs";
   import { querySQL } from "./query.mjs";
   
   /**
   * Execute the client functions.
   */
   async function main() {
     /** Write line protocol data to InfluxDB. */
     await writeLineProtocol();
     /** Query data from InfluxDB using SQL. */
     await querySQL();
   }
   
   main();
   

6. 在您的终端中，执行 index.mjs 以写入和查询 InfluxDB Clustered

   node index.mjs
   

1. 在您在 数据写入部分 中创建的 influxdb_csharp_client 目录中，创建一个名为 Query.cs 的新文件。

2. 在 Query.cs 中，输入以下示例代码

   // Query.cs
   
   using System;
   using System.Threading.Tasks;
   using InfluxDB3.Client;
   using InfluxDB3.Client.Query;
   
   namespace InfluxDBv3;
   
   public class Query
   {
     /**
       * Queries an InfluxDB database using the C# .NET client
       * library.
       **/
     public static async Task QuerySQL()
     {
       /** INFLUX_TOKEN is an environment variable you assigned to your
         * database READ token value.
         **/
       string? token = System.Environment
           .GetEnvironmentVariable("INFLUX_TOKEN");
   
       /**
         * Instantiate the InfluxDB client with credentials.
         **/
       using var client = new InfluxDBClient(
           "https://cluster-host.com", token: token, database: database);
   
       const string sql = @"
         SELECT time, room, temp, hum, co
         FROM home
         WHERE time >= '2022-01-01T08:00:00Z'
         AND time <= '2022-01-01T20:00:00Z'
       ";
   
       Console.WriteLine("{0,-30}{1,-15}{2,-15}{3,-15}{4,-15}",
           "time", "room", "co", "hum", "temp");
   
       await foreach (var row in client.Query(query: sql))
       {
         {
           /** 
             * Iterate over rows and print column values in table format.
             * Format the timestamp as sortable UTC format.
             */
           Console.WriteLine("{0,-30:u}{1,-15}{4,-15}{3,-15}{2,-15}",
               row[0], row[1], row[2], row[3], row[4]);
         }
       }
       Console.WriteLine();
     }
   }
   

   示例代码执行以下操作

   1. 导入以下类

      • System
      • System.Threading.Tasks;
      • InfluxDB3.Client;
      • InfluxDB3.Client.Query;

   2. 定义一个 Query 类，其中包含一个 QuerySQL() 方法，该方法执行以下操作

      1. 调用 new InfluxDBClient() 构造函数以实例化配置有 InfluxDB 凭据的客户端。

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

      2. 为 SQL 查询定义一个字符串变量。

      3. 调用 InfluxDBClient.Query() 方法以发送带有 SQL 字符串的查询请求。Query() 从响应流返回成批的行，形式为二维数组 - 行数组，其中每一行都是值数组。

      4. 迭代行并将数据以表格格式打印到 stdout。

3. 在您的编辑器中，打开您在 数据写入部分 中创建的 Program.cs 文件，并插入代码以调用 Query() 函数 - 例如

   // Program.cs
   
   using System;
   using System.Threading.Tasks;
   
   namespace InfluxDBv3;
   
   public class Program
   {
     public static async Task Main()
     {
       await Write.WriteLineProtocol();
       await Query.QuerySQL();
     }
   }
   

4. 要构建和执行程序并查询您的 InfluxDB 集群，请在您的终端中输入以下命令

   dotnet run
   

本教程假设使用 Maven 版本 3.9，Java 版本 >= 15，以及在 数据写入部分 中创建的 influxdb_java_client Maven 项目。

1. 在您的终端或编辑器中，更改为您在 数据写入部分 中创建的 influxdb_java_client 目录。

2. 在 src/main/java/com/influxdbv3 目录内部，创建一个名为 Query.java 的新文件。

3. 在 Query.java 中，输入以下示例代码

   // Query.java
   package com.influxdbv3;
   
   import com.influxdb.v3.client.InfluxDBClient;
   import java.util.stream.Stream;
   
   /**
     * Queries an InfluxDB database using the Java client
     * library.
     **/
   public final class Query {
   
       private Query() {
           //not called
       }
   
       /**
       * @throws Exception
       */
       public static void querySQL() throws Exception {
           /**
           * Query using SQL.
           */
   
           /** Set InfluxDB credentials. **/
           final String host = "https://cluster-host.com";
           final String database = "get-started";
   
           /** INFLUX_TOKEN is an environment variable you assigned to your
             * database READ token value.
             **/
           final char[] token = (System.getenv("INFLUX_TOKEN")).
           toCharArray();
   
           try (InfluxDBClient client = InfluxDBClient.getInstance(host,
           token, database)) {
               String sql =
                   """
                   SELECT time, room, temp, hum, co
                   FROM home
                   WHERE time >= '2022-01-01T08:00:00Z'
                   AND time <= '2022-01-01T20:00:00Z'""";
   
               String layoutHead = "| %-16s | %-12s | %-6s | %-6s | %-6s |%n";
               System.out.printf(
               "--------------------------------------------------------%n");
               System.out.printf(layoutHead,
               "time", "room", "co", "hum", "temp");
               System.out.printf(
               "--------------------------------------------------------%n");
               String layout = "| %-16s | %-12s | %-6s | %.1f | %.1f |%n";
   
               try (Stream<Object[]> stream = client.query(sql)) {
                   stream.forEach(row -> 
                     System.out.printf(layout,
                     row[0], row[1], row[4], row[3], row[2])
                   );
               }
           }
       }
   }
   

   示例代码执行以下操作

   1. 分配 com.influxdbv3 包名称（Maven groupId）。

   2. 导入以下类

      • com.influxdb.v3.client.InfluxDBClient
      • java.util.stream.Stream

   3. 定义一个 Query 类，其中包含一个 querySQL() 方法，该方法执行以下操作

      1. 调用 InfluxDBClient.getInstance() 以实例化配置有 InfluxDB 凭据的客户端。

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

      2. 为 SQL 查询定义一个字符串变量 (sql)。

      3. 为标题和数据行定义 Markdown 表格格式布局。

      4. 调用 InfluxDBClient.query() 方法以发送带有 SQL 字符串的查询请求。query() 返回行流。

      5. 迭代行并将数据以指定的布局打印到 stdout。

4. 在您的编辑器中，打开 src/main/java/com/influxdbv3/App.java 文件，并将内容替换为以下示例代码

   // App.java
   
   package com.influxdbv3;
   
   /**
   * Execute the client functions.
   *
   */
   public class App {
   
       /**
       * @param args
       * @throws Exception
       */
       public static void main(final String[] args) throws Exception {
           // Write data to InfluxDB 3.
           Write.writeLineProtocol();
           // Run the SQL query.
           Query.querySQL();
       }
   }
   

   • App、Write 和 Query 类属于 com.influxdbv3 包（您的项目 groupId）。
   • App 定义一个 main() 函数，该函数调用 Write.writeLineProtocol() 和 Query.querySQL()。

5. 在您的终端或编辑器中，使用 Maven 安装依赖项并编译项目代码 - 例如

   mvn compile
   

6. 为您的环境设置 --add-opens=java.base/java.nio=ALL-UNNAMED Java 选项。Apache Arrow Flight 库需要此设置才能访问 java.nio API 包。

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

   Linux/MacOS

   export MAVEN_OPTS="--add-opens=java.base/java.nio=ALL-UNNAMED"
   

   Windows PowerShell

   $env:MAVEN_OPTS="--add-opens=java.base/java.nio=ALL-UNNAMED"
   

7. 要运行应用程序以写入和查询 InfluxDB Clustered，请执行 App.main() - 例如，使用 Maven

   mvn exec:java -Dexec.mainClass="com.influxdbv3.App"