云数据库 SelectDB 版：通过Arrow Flight SQL连接云数据库SelectDB版实例

安装依赖

使用 pip 安装两个包：通用 ADBC 管理库 + Flight SQL 驱动。

pip install adbc_driver_manager adbc_driver_flightsql

连接并查询数据

下面的脚本连接 SelectDB 并将查询结果转为 Pandas DataFrame：

import adbc_driver_manager
import adbc_driver_flightsql.dbapi as flight_sql

# 用 adbc_driver_manager.DatabaseOptions 提供账号密码
conn = flight_sql.connect(
    uri="grpc://<VPC地址>:<Arrow Flight SQL端口>",
    db_kwargs={
        adbc_driver_manager.DatabaseOptions.USERNAME.value: "<数据库账号>",
        adbc_driver_manager.DatabaseOptions.PASSWORD.value: "<数据库密码>",
    },
)

cursor = conn.cursor()
cursor.execute("SELECT * FROM demo.example_table LIMIT 1000")

# fetchallarrow() 返回 Arrow Table，零拷贝转 Pandas
arrow_table = cursor.fetchallarrow()
df = arrow_table.to_pandas()
print(f"got {len(df)} rows")
print(df.head())

cursor.close()
conn.close()

关于 DDL / DML / 管理语句

CREATE TABLE / INSERT / SET / SHOW VARIABLES 等命令可以通过 cursor.execute() 提交并在服务端正常生效，但获取返回状态需要 ADBC 客户端连到 FE 节点拉取 StatusResult，部分场景下会出现 i/o timeout。建议把这一类操作改用 MySQL 9030 协议（pymysql、JDBC mysql 驱动）执行；Arrow Flight SQL 通道专门用于批量读取查询结果。

添加 Maven 依赖

在 pom.xml 中新增依赖。建议把版本号定义在 properties 中，便于统一升级：

<properties>
    <arrow.version>17.0.0</arrow.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.arrow</groupId>
        <artifactId>flight-sql-jdbc-core</artifactId>
        <version>${arrow.version}</version>
    </dependency>
</dependencies>

连接并查询

JDBC URL 协议头为 jdbc:arrow-flight-sql，其余用法与 jdbc:mysql 接近。返回的 ResultSet 由 Arrow Table 包装而成：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class ArrowFlightSqlExample {
    public static void main(String[] args) throws Exception {
        String url = "jdbc:arrow-flight-sql://<VPC地址>:<Arrow Flight SQL端口>"
                   + "?useServerPrepStmts=false"
                   + "&useSSL=false"
                   + "&useEncryption=false";
        String user = "<数据库账号>";
        String password = "<数据库密码>";

        Class.forName("org.apache.arrow.driver.jdbc.ArrowFlightJdbcDriver");

        try (Connection conn = DriverManager.getConnection(url, user, password);
             Statement stmt = conn.createStatement()) {

            ResultSet rs = stmt.executeQuery("SELECT * FROM demo.example_table LIMIT 100");
            while (rs.next()) {
                System.out.println(rs.getInt(1) + " | " + rs.getString(2));
            }
            rs.close();
        }
    }
}

添加 Maven 依赖

<properties>
    <adbc.version>0.15.0</adbc.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-core</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-driver-manager</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-driver-flight-sql</artifactId>
        <version>${adbc.version}</version>
    </dependency>
</dependencies>

连接并迭代结果

import org.apache.arrow.adbc.core.AdbcConnection;
import org.apache.arrow.adbc.core.AdbcDatabase;
import org.apache.arrow.adbc.core.AdbcDriver;
import org.apache.arrow.adbc.core.AdbcStatement;
import org.apache.arrow.adbc.driver.flightsql.FlightSqlDriver;
import org.apache.arrow.flight.Location;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowReader;

import java.util.HashMap;
import java.util.Map;

public class ArrowFlightAdbcExample {
    public static void main(String[] args) throws Exception {
        BufferAllocator allocator = new RootAllocator();
        FlightSqlDriver driver = new FlightSqlDriver(allocator);

        Map<String, Object> parameters = new HashMap<>();
        AdbcDriver.PARAM_URI.set(parameters,
            Location.forGrpcInsecure("<VPC地址>", <Arrow Flight SQL端口>).getUri().toString());
        AdbcDriver.PARAM_USERNAME.set(parameters, "<数据库账号>");
        AdbcDriver.PARAM_PASSWORD.set(parameters, "<数据库密码>");

        try (AdbcDatabase db = driver.open(parameters);
             AdbcConnection connection = db.connect();
             AdbcStatement stmt = connection.createStatement()) {

            stmt.setSqlQuery("SELECT * FROM demo.example_table LIMIT 100");
            try (AdbcStatement.QueryResult result = stmt.executeQuery();
                 ArrowReader reader = result.getReader()) {

                while (reader.loadNextBatch()) {
                    VectorSchemaRoot batch = reader.getVectorSchemaRoot();
                    System.out.println("rows in batch: " + batch.getRowCount());
                    // 在这里把 batch 接入 Spark / Flink / 自定义计算
                }
            }
        }
        allocator.close();
    }
}