Go Flight 客户端

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

Flight SQL 客户端

使用 Flight SQL 的示例查询

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

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 := "cluster-host.com: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集群的主机名和端口（《:443》）（无协议）
         • database：要查询的InfluxDB集群数据库的名称
         • token：具有指定数据库读取权限的数据库令牌。 出于安全考虑，我们建议将其设置为环境变量，而不是包含原始令牌字符串。

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

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

      4. 将以下InfluxDB凭证作为键值对追加到出站上下文中

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

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

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

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

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

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

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

   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 Client文档。