Go Flight 客户端

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

Flight SQL 客户端

使用 Flight SQL 的示例查询

以下示例演示了如何使用 Arrow Flight SQL Go 客户端查询 InfluxDB 3 核心数据库。

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 核心集群主机名和端口 (:443) （不含协议）
         • database：要查询的 InfluxDB 3 核心数据库的名称。
         • token：一个具有指定数据库读取权限的 数据库令牌。出于安全原因，我们建议将其设置为环境变量，而不是包含原始令牌字符串。

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

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

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

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

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

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

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

      8. 创建一个 reader 来读取端点返回的 Arrow 表，并将结果以 JSON 格式打印出来。

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

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

   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 客户端文档。