开始查询数据

使用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 也会安装用于处理从查询返回的 Arrow 数据的 pyarrow 库。

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 集群与您的 Python 代码集成。 客户端库支持将数据写入 InfluxDB 和使用 SQL 或 InfluxQL 查询数据。

以下步骤包括设置一个Python虚拟环境，已经在开始编写数据中涵盖过。

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集群数据库的名称

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

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

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

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

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

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. 用以下参数实例化 influx.Client 以获取 InfluxDB 凭据：

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

      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://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集群数据库的名称

      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 集群：

   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集群数据库的名称
         • token: 一个数据库令牌，具有对指定数据库的读权限。 将其存储在秘密存储或环境变量中，以避免暴露原始令牌字符串。

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

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

      4. 遍历行并将数据以表格格式打印到标准输出。

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集群数据库的名称
         • token: 一个 数据库令牌，具有对指定数据库的读取权限。 将其存储在秘密存储或环境变量中，以避免暴露原始令牌字符串。

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

      3. 定义了一个用于标题和数据行的Markdown表格格式布局。

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

      5. 遍历行并以指定的布局将数据打印到标准输出。

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集群，请执行 App.main() - 例如，使用Maven：

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