Go Flight 客户端

Apache Arrow for Go 集成了 Go 脚本和应用程序，用于查询存储在 InfluxDB 中的数据。

Flight SQL 客户端

使用 Flight SQL 的查询示例

以下示例展示了如何使用 Arrow Flight SQL Go 客户端查询 InfluxDB 3 Core 数据库

1. 在您的编辑器中，打开一个名为 query.go 的新文件，并输入以下示例代码

   package main
   
   import (
     "context"
     "crypto/x509"
     "encoding/json"
     "fmt"
     "os"
   
     "github.com/apache/arrow/go/v14/arrow/flight/flightsql"
     "google.golang.org/grpc"
     "google.golang.org/grpc/credentials"
     "google.golang.org/grpc/metadata"
   )
   
   func dbQuery(ctx context.Context) error {
     url := "localhost:8181:443"
   
     // INFLUX_TOKEN is an environment variable you created for your database READ token
     token := os.Getenv("INFLUX_TOKEN")
     database := "get-started"
   
     // Create a gRPC transport
     pool, err := x509.SystemCertPool()
     if err != nil {
       return fmt.Errorf("x509: %s", err)
     }
     transport := grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(pool, ""))
     opts := []grpc.DialOption{
       transport,
     }
   
     // Create query client
     client, err := flightsql.NewClient(url, nil, nil, opts...)
     if err != nil {
       return fmt.Errorf("flightsql: %s", err)
     }
   
     ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+token)
     ctx = metadata.AppendToOutgoingContext(ctx, "database", database)
   
     // Execute query
     query := `SELECT
       *
     FROM
       home
     WHERE
       time >= '2022-01-01T08:00:00Z'
       AND time <= '2022-01-01T20:00:00Z'`
   
     info, err := client.Execute(ctx, query)
     if err != nil {
       return fmt.Errorf("flightsql flight info: %s", err)
     }
     reader, err := client.DoGet(ctx, info.Endpoint[0].Ticket)
     if err != nil {
       return fmt.Errorf("flightsql do get: %s", err)
     }
   
     // Print results as JSON
     for reader.Next() {
       record := reader.Record()
       b, err := json.MarshalIndent(record, "", "  ")
       if err != nil {
         return err
       }
       fmt.Println("RECORD BATCH")
       fmt.Println(string(b))
   
       if err := reader.Err(); err != nil {
         return fmt.Errorf("flightsql reader: %s", err)
       }
     }
   
     return nil
   }
   
   func main() {
     if err := dbQuery(context.Background()); err != nil {
       fmt.Fprintf(os.Stderr, "error: %v\n", err)
       os.Exit(1)
     }
   }
   

   此示例执行以下操作

   1. 导入以下软件包

      • context
      • crypto/x509
      • encoding/json
      • fmt
      • os
      • github.com/apache/arrow/go/v14/arrow/flight/flightsql
      • google.golang.org/grpc
      • google.golang.org/grpc/credentials
      • google.golang.org/grpc/metadata

   2. 创建一个执行以下操作的 dbQuery 函数

      1. 定义 InfluxDB 凭据的变量。

         • url: InfluxDB 3 Core 集群主机名和端口 (:443) (无协议)
         • database: 要查询的 InfluxDB 3 Core 数据库的名称
         • token: 具有对指定数据库读取权限的数据库令牌。 出于安全原因，我们建议将其设置为环境变量，而不是包含原始令牌字符串。

      2. 定义一个 opts 选项列表，其中包含用于通过 gRPC+TLS 协议与 InfluxDB 通信的 gRPC 传输。

      3. 调用 flightsql.NewClient() 方法，并使用 url 和 opts 创建一个新的 Flight SQL 客户端。

      4. 将以下 InfluxDB 凭据作为键值对附加到传出的上下文中

         • authorization: Bearer <INFLUX_TOKEN>
         • database: 数据库名称

      5. 定义要执行的 SQL 查询。

      6. 调用 client.execute() 方法以发送查询请求。

      7. 使用查询响应中的 ticket 调用 client.doGet() 方法，以从端点检索结果数据。

      8. 创建一个读取器以读取端点返回的 Arrow 表，并将结果打印为 JSON。

   3. 创建一个执行 dbQuery 函数的 main 模块函数。

2. 输入以下命令以安装所有必要的软件包并运行程序以查询 InfluxDB 3 Core

   go get ./...
   go run ./query.go
   

RECORD BATCH
[
  {
    "co": 0,
    "hum": 35.9,
    "room": "Kitchen",
    "temp": 21,
    "time": "2022-01-01 08:00:00"
  },
  {
    "co": 0,
    "hum": 36.2,
    "room": "Kitchen",
    "temp": 23,
    "time": "2022-01-01 09:00:00"
  },
  {
    "co": 0,
    "hum": 36.1,
    "room": "Kitchen",
    "temp": 22.7,
    "time": "2022-01-01 10:00:00"
  },
  {
    "co": 0,
    "hum": 36,
    "room": "Kitchen",
    "temp": 22.4,
    "time": "2022-01-01 11:00:00"
  },
  {
    "co": 0,
    "hum": 36,
    "room": "Kitchen",
    "temp": 22.5,
    "time": "2022-01-01 12:00:00"
  },
  ...
]

有关更多信息，请参阅 Go Arrow Flight 客户端文档。