Java 航班 SQL 包

Apache Arrow Flight SQL for Java 集成了Java应用程序，通过RPC和SQL查询并检索来自Flight数据库服务器的数据。

使用 InfluxDB 3 客户端库

我们推荐使用influxdb3-java Java客户端库将InfluxDB 3与您的Java应用程序代码集成。

InfluxDB 3 客户端库封装了 Apache Arrow Flight 客户端并提供了方便的方法用于写入、查询和处理存储在 InfluxDB Cloud Serverless 中的数据。客户端库可以使用 SQL 或 InfluxQL 进行查询。

开始使用Java Flight SQL客户端查询InfluxDB
故障排除箭头飞行请求

开始使用Java Flight SQL客户端查询InfluxDB

编写一个 Java 类，用于连接到 InfluxDB Cloud Serverless 的 Flight SQL 客户端，执行 SQL 查询，并检索存储在 InfluxDB Cloud Serverless 存储桶中的数据。

示例使用了Apache Arrow Java 实现 (org.apache.arrow)来与像 InfluxDB 3 这样的 Flight 数据库服务器进行交互。

org.apache.arrow: 提供了用于将Java应用程序与Apache Arrow数据和协议集成的类和方法。
org.apache.arrow.flight.sql: 提供与使用 Arrow Flight RPC 和 Flight SQL 的 Flight 数据库服务器进行交互的类和方法。

设置 InfluxDB
安装先决条件
创建 FlightQuery 类
创建查询客户端
执行查询
检索和处理Arrow数据

要克隆或下载您可以使用 Docker 运行的示例应用程序，请查看 GitHub 上的 InfluxCommunity/ArrowFlightClient_Query_Examples repository。

设置 InfluxDB

要配置应用程序以查询 InfluxDB Cloud Serverless，您需要以下 InfluxDB 资源：

InfluxDB Cloud 无服务器桶
具有对数据桶的读取权限的InfluxDB Cloud无服务器API令牌

如果您还没有API令牌和存储桶，请查看如何设置InfluxDB。如果您还没有要查询的数据，请查看如何写入数据到存储桶。

安装前提条件

以下使用Docker和Maven构建并运行Java应用程序，以避免特定于平台的依赖问题。

这个示例 Dockerfile 在 Docker 容器中安装兼容版本的 Maven 和 Java JDK，然后运行 Maven 命令以下载依赖项并编译应用程序。

按照说明下载并安装适合您系统的Docker：

macOS: 安装Docker用于macOS
Linux: 为Linux安装Docker

查看 Dockerfile

查看 Maven pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.influxdb</groupId>
  <artifactId>examples</artifactId>
  <version>1.0-SNAPSHOT</version>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-jar-plugin</artifactId>
        <version>3.2.0</version>
        <configuration>
          <archive>
            <manifest>
              <addClasspath>true</addClasspath>
              <classpathPrefix>lib/</classpathPrefix>
              <mainClass>com.influxdb.examples.FlightExamples</mainClass>
            </manifest>
          </archive>
        </configuration>
      </plugin>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-help-plugin</artifactId>
        <version>3.2.0</version>
      </plugin>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>3.4.1</version>
        <executions>
          <execution>
            <goals>
              <goal>shade</goal>
            </goals>
            <configuration>
              <shadedArtifactAttached>true</shadedArtifactAttached>
            </configuration>
          </execution>
        </executions>
        <configuration>
          <minimizeJar>false</minimizeJar>
          <filters>
            <filter>
              <artifact>*:*</artifact>
              <excludes>
                <exclude>META-INF/*.SF</exclude>
                <exclude>META-INF/*.DSA</exclude>
                <exclude>META-INF/*.RSA</exclude>
              </excludes>
            </filter>
          </filters>
        </configuration>
      </plugin>
    </plugins>
  </build>

  <properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.arrow</groupId>
      <artifactId>flight-sql</artifactId>
      <version>11.0.0</version>
    </dependency>
    <dependency>
      <groupId>io.netty</groupId>
      <artifactId>netty-all</artifactId>
      <version>4.1.74.Final</version>
    </dependency>
    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-simple</artifactId>
      <version>1.7.30</version>
    </dependency>
  </dependencies>

</project>

创建 FlightQuery 类

查看 FlightQuery.java

package com.influxdb.examples;

import org.apache.arrow.flight.auth2.BearerCredentialWriter;
import org.apache.arrow.flight.CallHeaders;
import org.apache.arrow.flight.CallStatus;
import org.apache.arrow.flight.grpc.CredentialCallOption;
import org.apache.arrow.flight.Location;
import org.apache.arrow.flight.FlightClient;
import org.apache.arrow.flight.FlightClientMiddleware;
import org.apache.arrow.flight.FlightInfo;
import org.apache.arrow.flight.FlightStream;
import org.apache.arrow.flight.sql.FlightSqlClient;
import org.apache.arrow.flight.Ticket;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;

public class FlightQuery {

    /* Get server credentials from environment variables */
    public static final String DATABASE_NAME = System.getenv("DATABASE_NAME");
    public static final String HOST = System.getenv("HOST");
    public static final String TOKEN = System.getenv("TOKEN");

    public static void main() {

        System.out.println("Query InfluxDB with the Java Flight SQL Client");

        // Create an interceptor that injects header metadata (database name) in every request.
        FlightClientMiddleware.Factory f = info -> new FlightClientMiddleware() {
            @Override
            public void onBeforeSendingHeaders(CallHeaders outgoingHeaders) {
                outgoingHeaders.insert("database", DATABASE_NAME);
            }

            @Override
            public void onHeadersReceived(CallHeaders incomingHeaders) {

            }

            @Override
            public void onCallCompleted(CallStatus status) {

            }
        };

        // Create a gRPC+TLS channel URI with HOST and port 443.
        Location location = Location.forGrpcTls(HOST, 443);

        // Set the allowed memory.
        BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE);

        // Create a client with the allocator and gRPC channel.
        FlightClient client = FlightClient.builder(allocator, location)
                .intercept(f)
                .build();
        System.out.println("client" + client);

        FlightSqlClient sqlClient = new FlightSqlClient(client);
        System.out.println("sqlClient: " + sqlClient);

        // Define the SQL query to execute.
        String query = "SELECT * FROM home";
        
        /*  Construct a bearer credential using TOKEN.
            Construct a credentials option using the bearer credential.
        */
        CredentialCallOption auth = new CredentialCallOption(new BearerCredentialWriter(TOKEN));

        /*  Execute the query.
            If successful, execute returns a FlightInfo object that contains metadata
            and an endpoints list.
            Each endpoint contains the following:
                - A list of addresses where you can retrieve the data.
                - A `ticket` value that identifies the data to retrieve.
        */
        FlightInfo flightInfo = sqlClient.execute(query, auth);

        // Extract the Flight ticket from the response.
        Ticket ticket = flightInfo.getEndpoints().get(0).getTicket();
        
        // Pass the ticket to request the Arrow stream data from the endpoint.
        final FlightStream stream = sqlClient.getStream(ticket, auth);

        // Process all the Arrow stream data.
        while (stream.next()) {
            try {
                // Get the current vector data from the stream.
                final VectorSchemaRoot root = stream.getRoot();
                System.out.println(root.contentToTSVString());
            } catch (Exception e) {
                // Handle exceptions.
                System.out.println("Error executing FlightSqlClient: " + e.getMessage());
            }
        }
        try {
            // Close the stream and release resources.
            stream.close();
        } catch (Exception e) {
            // Handle exceptions.
            System.out.println("Error closing stream: " + e.getMessage());
        }

        try {
            // Close the client
            sqlClient.close();
        } catch (Exception e) {
            // Handle exceptions.
            System.out.println("Error closing client: " + e.getMessage());
        }
    }
}

在你的 <PROJECT_ROOT>/src/main/java 目录中，创建 com/influxdb/examples 子目录，用于 com.influxdb.examples 包。

在前一步的 examples 目录中，创建 FlightQuery.java 类文件。你应该有以下目录结构：

PROJECT_ROOT
└──src
   └──main
      └──java
         └──com
            └──influxdb
               └──examples
                  └──FlightQuery.java

在 FlightQuery.java 中：
1. 添加包名称：
```
package com.influxdb.examples;
```
2. 为以下包添加 import 语句。您将在后续步骤中使用这些包中的类和方法：
  - org.apache.arrow.flight.auth2.BearerCredentialWriter
  - org.apache.arrow.flight.CallHeaders
  - org.apache.arrow.flight.CallStatus
  - org.apache.arrow.flight.grpc.CredentialCallOption
  - org.apache.arrow.flight.Location
  - org.apache.arrow.flight.FlightClient
  - org.apache.arrow.flight.FlightClientMiddleware
  - org.apache.arrow.flight.FlightInfo
  - org.apache.arrow.flight.FlightStream
  - org.apache.arrow.flight.sql.FlightSqlClient
  - org.apache.arrow.flight.Ticket
  - org.apache.arrow.memory.BufferAllocator
  - org.apache.arrow.memory.RootAllocator
  - org.apache.arrow.vector.VectorSchemaRoot
3. 创建一个 FlightQuery 类。
4. 在FlightQuery类中：
  1. 为服务器凭据定义常量。
    - DATABASE_NAME
    - HOST
    - TOKEN
    这个示例 Dockerfile 定义了这些凭证的环境变量。
  2. 创建一个 main() 方法。

创建查询客户端

在 FlightQuery.main() 方法中，执行以下操作以创建一个可以连接到 HOST 和 DATABASE_NAME 的SQL客户端：

构建一个 gRPC+TLS 通道 URI，其中包含 HOST 和端口 443，用于通过 gRPC 服务器进行 TLS 通信。
实例化 FlightClientMiddleware 并定义一个事件回调，它插入以下飞行请求元数据头属性：
```
"database": "DATABASE_NAME"
```
实例化一个 BufferAllocator，用于设置客户端允许的内存。
创建一个 FlightClient，使用分配器和 gRPC 通道。
实例化一个 FlightSqlClient，它包装了 FlightClient 实例。

执行查询

在 FlightQuery.main 方法中：

实例化一个 CredentialCallOption，使用 TOKEN 作为 持有者 凭证。结果是一个凭证对象，您将在每个请求中传递给服务器。
定义一个包含要执行的SQL查询的字符串–例如：
```
String query = "SELECT * FROM home";
```
使用 SQL 查询和 CredentialCallOption 调用 FlightSqlClient.execute 方法。
如果成功，FlightSqlClient.execute 方法将返回一个 FlightInfo 对象，其中包含元数据和一个 endpoints: [...] 列表。每个端点包含以下内容：
- 您可以获取数据的地址列表。
- 一个 ticket 值，用于标识要检索的数据。
从响应中提取票据。

检索和处理箭头数据

在FlightQuery.main()方法中，执行以下操作以检索在FlightInfo响应中描述的数据流：

调用 FlightSqlClient.getStream 方法，使用 ticket 和 CredentialCallOption 来获取 Arrow stream。
调用 FlightStream.getRoot 方法从流中获取当前的矢量数据。
处理数据并处理异常。该示例将向量数据转换为制表符分隔的值，并将结果打印到 System.out。
有关使用Java处理Arrow数据的更多示例，请参见Apache Arrow Java Cookbook。
最后，关闭流和客户端。

运行应用程序

按照以下步骤使用Docker构建和运行应用程序：

将 Dockerfile 和 pom.xml 复制到您的项目根目录。
在您的项目根目录中打开一个终端。
在你的终端中，运行 docker build 命令并传递 --build-arg 标志用于服务器凭据：
- DATABASE_NAME: 你的 InfluxDB Cloud Serverless bucket
- HOST: 您的 InfluxDB Cloud Serverless 区域主机名 (不带“https://”的 URL)
- TOKEN: 您的 InfluxDB Cloud Serverless API 令牌具有对存储桶的读取权限
```
docker build \
--build-arg DATABASE_NAME=INFLUX_BUCKET \
--build-arg HOST=cloud2.influxdata.com \
--build-arg TOKEN=INFLUX_TOKEN \
-t javaflight .
```
该命令构建一个名为 javaflight 的Docker镜像。
要在新的Docker容器中运行应用程序，请输入以下命令：
```
docker run javaflight
```
输出是TSV格式的查询数据。

故障排除箭头飞行请求

有关 Arrow Flight 错误响应代码的列表，请参见 Arrow Flight RPC documentation。

Java 航班 SQL 包

使用 InfluxDB 3 客户端库

开始使用Java Flight SQL客户端查询InfluxDB

设置 InfluxDB

安装前提条件

创建 FlightQuery 类

创建查询客户端

执行查询

检索和处理箭头数据

运行应用程序

故障排除箭头飞行请求

Flux的未来

InfluxDB 3 开源版本现已公开Alpha测试

Java 航班 SQL 包

使用 InfluxDB 3 客户端库

开始使用Java Flight SQL客户端查询InfluxDB

设置 InfluxDB

安装前提条件

创建 FlightQuery 类

创建查询客户端

执行查询

检索和处理箭头数据

运行应用程序

故障排除箭头飞行请求

相关

选择您的 InfluxDB 云无服务器区域

AWS

感谢您的反馈!

Flux的未来

InfluxDB 3 开源版本现已公开Alpha测试

InfluxDB 云端无服务器