开始查询数据

使用 SQL 和 influx3 CLI 查询 InfluxDB 3。

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

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: 一个具有 读取 访问权限的 InfluxDB API token，用于 get-started 桶
   • 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 Cloud Serverless与您的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"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云无服务器区域主机名 （不包括 https:// 协议或末尾斜杠）
   • database: 要查询的InfluxDB Cloud Serverless bucket的名称
   • token: 一个API令牌，具有对指定存储桶的读取权限。将其存储在秘密存储或环境变量中，以避免暴露原始令牌字符串。

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

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

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

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

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

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/InfluxCommunity/influxdb3-go/v2/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 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 Cloud Serverless 区域 URL
         • Database: 您的 InfluxDB Cloud Serverless 存储桶的名称
         • Token: 一个 API 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://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 token 具有对您想要查询的存储桶的 读取 权限。 将此存储在秘密存储或环境变量中，以避免暴露原始的token字符串。

   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. 定义一个 Query 类，其中包含一个 QuerySQL() 方法，该方法执行以下操作：

      1. 调用 new InfluxDBClient() 构造函数来实例化一个配置了 InfluxDB 凭证的客户端。

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

      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 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. 定义一个 Query 类，包含一个 querySQL() 方法，该方法执行以下操作：

      1. 调用 InfluxDBClient.getInstance() 来实例化一个配置了 InfluxDB 凭据的客户端。

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

      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 package。

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

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

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