开始查询数据

使用 SQL 和 influx3 CLI 查询 InfluxDB v3。

以下步骤包括设置 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 create \
     --name="config-serverless" \
     --database="get-started" \
     --host="cloud2.influxdata.com" \
     --token="API_TOKEN" \
     --org="ORG_ID"
   

   替换以下内容

   • API_TOKEN：具有对 get-started 存储桶 读取 访问权限的 InfluxDB API 令牌
   • ORG_ID：InfluxDB 组织 ID

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 云无服务器集成到您的 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客户端库模块，并安装了用于处理查询返回的Arrow数据的pyarrow包。
      • 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"cloud2.influxdata.com",
       token=f"API_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 Cloud Serverless区域主机名（不包含https://协议或尾随斜杠）
   • database：要查询的InfluxDB Cloud Serverless bucket的名称
   • token：具有对指定bucket 读取访问权限的API token。将其存储在秘密存储或环境变量中，以避免暴露原始token字符串。

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

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

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

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

7. 调用 print() 方法将 markdown 表格打印到标准输出。

8. 在您的终端中，输入以下命令以运行程序并查询 InfluxDB Cloud Serverless

   python query.py
   

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

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

   package main
   
   import (
     "context"
     "fmt"
     "io"
     "os"
     "time"
     "text/tabwriter"
   
     "github.com/apache/arrow/go/v13/arrow"
     "github.com/InfluxCommunity/influxdb3-go/influxdb3"
   )
   
   func Query() error {
   
      // INFLUX_TOKEN is an environment variable you created
      // for your API read token.
     token := os.Getenv("INFLUX_TOKEN")
   
     // Instantiate the client.
     client, err := influxdb3.New(influxdb3.ClientConfig{
       Host:     "https://cloud2.influxdata.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 arrow and time packages to format unix timestamp
       // as a time with timezone layout (RFC3339).
       time := (row["time"].(arrow.Timestamp)).
         ToTime(arrow.TimeUnit(arrow.Nanosecond)).
         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/apache/arrow/go/v13/arrow
      • github.com/InfluxCommunity/influxdb3-go/influxdb3

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

      1. 使用 InfluxDB 凭据实例化 influx.Client。

         • Host：您的 InfluxDB Cloud Serverless 区域 URL
         • Database：您的 InfluxDB Cloud Serverless 存储桶名称
         • Token：一个具有对指定存储桶读取权限的 API 令牌。 请将其存储在密钥存储或环境变量中，以避免泄露原始令牌字符串。

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

      3. 定义一个用于 SQL 查询的字符串变量。

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

      5. 遍历行，将时间戳格式化为 RFC3339 时间戳，并以表格格式将数据打印到标准输出。

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://cloud2.influxdata.com";
   const database = 'get-started';
   /**
   * INFLUX_TOKEN is an environment variable you assigned to your
   * API 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 Cloud Serverless 区域 URL
      • token：一个具有对您要查询的存储桶 读取 权限的 API 令牌。 请将其存储在密钥存储或环境变量中，以避免泄露原始令牌字符串。

   3. 定义一个用于 SQL 查询的字符串变量（sql）。

   4. 定义一个对象（data），以列名为键，以数组值为行数据的存储。

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

      • sql：要执行的查询
      • database：要查询的 InfluxDB Cloud Serverless 桶名称

      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 Cloud Serverless

   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 (bucket) using the C# .NET client
       * library.
       **/
     public static async Task QuerySQL()
     {
       /** INFLUX_TOKEN is an environment variable you assigned to your
         * API READ token value.
         **/
       string? token = System.Environment
           .GetEnvironmentVariable("INFLUX_TOKEN");
   
       /**
         * Instantiate the InfluxDB client with credentials.
         **/
       using var client = new InfluxDBClient(
           "https://cloud2.influxdata.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. 定义一个具有 QuerySQL() 方法的 Query 类，该方法执行以下操作

      1. 调用 new InfluxDBClient() 构造函数以使用 InfluxDB 凭据配置客户端。

         • host：您的 InfluxDB Cloud Serverless 区域 URL。
         • token：一个具有对指定桶读取权限的 API 令牌。将其存储在秘密存储或环境变量中，以避免暴露原始令牌字符串。
         • database：要查询的 InfluxDB Cloud Serverless 桶名称

      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 Cloud Serverless，请在您的终端中输入以下命令

   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 (bucket) 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://cloud2.influxdata.com";
           final String database = "get-started";
   
           /** INFLUX_TOKEN is an environment variable you assigned to your
             * API 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. 定义一个具有 querySQL() 方法的 Query 类，该方法执行以下操作

      1. 调用 InfluxDBClient.getInstance() 以使用 InfluxDB 凭据配置客户端。

         • host：您的 InfluxDB Cloud Serverless 区域 URL
         • database：要写入的 InfluxDB Cloud Serverless 桶的名称
         • token：具有对指定bucket 读取访问权限的API token。将其存储在秘密存储或环境变量中，以避免暴露原始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 v3.
           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 云无服务器应用程序，请执行 App.main() - 例如，使用 Maven

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