使用 Go Flight SQL 客户端查询 InfluxDB Cloud

作者：Anais Dotis-Georgiou / 产品
2023 年 6 月 7 日

导航至

InfluxDB Cloud 3.0 是一个通用的时间序列数据库，构建于 Apache 生态系统之上。您可以使用 Apache Arrow Flight SQL 接口查询 InfluxDB Cloud，该接口为处理时间序列数据提供 SQL 支持。在本教程中，我们将逐步介绍使用 Go 和 Flight SQL 查询 InfluxDB Cloud 的过程。Go Flight SQL 客户端是 Apache Arrow Flight 的一部分，Apache Arrow Flight 是一个用于构建高性能数据服务的框架。它提供了一种通过 gRPC（一种现代高性能 RPC 框架）高效传输大型数据集的方法。通过此存储库，亲自试用使用 Go Flight SQL 客户端查询 InfluxDB Cloud。

要求和设置

本教程假定您已经拥有一个免费的 InfluxDB Cloud 帐户，并且您的机器上运行着 Docker。

最后，您需要创建或获取以下 InfluxDB 资源

  • 存储桶

  • 令牌

  • 您的组织（通常是您注册帐户时使用的电子邮件）

您还需要将数据写入您的 InfluxDB 帐户。最简单的方法是通过 UI 手动编写一些行协议。导航到加载数据 > 存储桶 > +添加数据 > 行协议 > 手动输入，选择您要写入的存储桶，然后将一个点写入 InfluxDB。例如，您可以编写 measurementName,tagKey=tagValue fieldKey=1.0。或者，如果您想要真实世界的行协议数据，请尝试 NOAA 空气传感器数据集。您还可以查看以下有关将数据写入 InfluxDB Cloud 3.0 的文档，了解将数据写入 InfluxDB Cloud 的其他方法。

代码演练

让我们将代码分解成更小的部分，以了解正在发生的事情。

  1. 导入所需的类：我们首先从 Apache Arrow Flight 和其他必要的库中导入所需的类。
    package main

import (
	"context"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"os"

	"github.com/apache/arrow/go/v12/arrow/flight/flightsql"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
	"os"
)
  2. 定义 main 函数以调用 dbQuery() 函数并处理错误。
    func main() {
	fmt.Println("An example using Go Flight SQL Client to query InfluxDB")

	if err := dbQuery(context.Background()); err != nil {
		fmt.Fprintf(os.Stderr, "error: %v\n", err)
		os.Exit(1)
	}
  3. 收集我们的身份验证环境变量，并在 dbQuery() 函数之后设置连接
    func dbQuery(ctx context.Context) error {
	url := os.Getenv("HOST")
	token := os.Getenv("TOKEN")
	bucket := os.Getenv("DATABASE_NAME")
…
// 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", bucket)
  4. 接下来，我们定义要执行的查询并执行它。使用 Execute 方法并传入 auth 和 query 以在服务器上执行查询。
    // Execute query
	query := `SELECT * FROM 'measurementName'`

	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)
	}
  5. 最后，我们将结果打印为 JSON。
    // 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)
		}
	}

使用 Go Flight SQL 查询 InfluxDB Cloud

要运行该示例，请首先克隆相应的存储库，导航到该目录，然后按照 README.md 中概述的以下步骤操作，或者

  1. 在您的终端中，设置以下环境变量。
    # Set environment variables

export INFLUX_DATABASE="your bucket" && \
export INFLUX_HOST="your host url i.e. us-east-1-1.aws.cloud2.influxdata.com" && \
export INFLUX_TOKEN="your token"
  2. 运行以下命令以使用 shell 脚本，该脚本构建一个名为 goflight 的映像
    sh ./influxdb-build.sh build
  3. 要启动应用程序，请在终端中运行 docker run "IMAGE_NAME"
    docker run goflight

资源和结论

请查看以下文档。它帮助我构建了这个示例，并且可以帮助您在查询 InfluxDB Cloud 的旅程中

  1. Arrow Flight 的参考文档

  2. InfluxDB Cloud 文档，关于 在 Python 中使用 Arrow Flight SQL 查询数据

  3. 一篇关于 InfluxDB、Flight SQL、Pandas 和 Jupyter Notebooks 教程 的博客文章

  4. 一篇关于 TL;DR InfluxDB 技术技巧：使用 Flight SQL 和 AWS Lambda 进行降采样 的博客文章

我希望这篇博客文章能启发您探索 InfluxDB Cloud 并利用 Flight SQL 从 InfluxDB 传输大型数据集，以便使用您选择的工具进行数据处理。如果您需要任何帮助，请使用我们的社区网站或 Slack 频道与我们联系。我很乐意了解您尝试实现的目标以及您希望 InfluxDB 拥有的功能。