跳到主要内容
版本:2.3.2

Arrow Flight SQL

Arrow Flight SQL 简介

Arrow Flight SQL 是一种使用 Arrow 内存格式和 Flight RPC 框架与 SQL 数据库交互的协议。

目前我们支持Arrow Flight SQL 客户端的环境有:

Arrow Flight SQL 优势

  1. 功能强大。功能与JDBC和ODBC等API类似,包括执行查询,创建准备好的语句。
  2. 安全。Flight,支持开箱即用的加密和身份验证等功能。
  3. 性能。与实现Arrow Flight 的客户端服务端通信,无需进行数据转化,同时允许进一步优化,如并行数据访问。

虽然它可以直接用于数据库访问,但它不能直接替代 JDBC/ODBC。 但是,Flight SQL 可以用作具体的有线协议/驱动程序实现,支持 JDBC/ODBC 驱动程序,并减少数据库的实现负担。

Arrow Flight SQL 查询流程

客户端使用arrow flight sql 客户端与数据库连接,查询数据,执行SQL的流程大致如下。

  1. 创建FlightSql客户端
  2. 验证用户名,密码
  3. 执行SQL,获取FlightInfo结构体
  4. 通过FlightInfo结构体中的FlightEndPoint获取到FlightData数据流

FlightInfo中包含有关数据所在位置的详细信息, 客户端可以从适当的服务器获取数据。 服务器信息被编码为 FlightInfo 中的一系列 FlightEndpoint 消息。 每个Endpoint代表包含响应数据子集的某个位置。

一个FlightEndpoint包含一个服务器地址列表, 一个Ticket, 一个服务器用来识别请求数据的二进制Token。 FlightEndPoint 没有定义顺序,如果数据集是排序的, 只会在一个FlightEndPoint中返回数据。

流程图如下

流程图

不同客户端的使用方式

提示

本章节分别介绍不同客户端的使用方式。

  • 安装Apache Arrow

    你可以去官方文档找到详细的安装教程 在Mac系统下,使用brew命令就可以简单安装了。

    brew install apache-arrow
    brew install apache-arrow-glib
  • 配置CMakeLists.txt

    cmake_minimum_required(VERSION 3.24)
    project(arrow_flight_cpp)

    set(CMAKE_CXX_STANDARD 20)

    find_package(Arrow REQUIRED)
    find_package(ArrowFlight REQUIRED)
    find_package(ArrowFlightSql REQUIRED)

    include_directories(${ARROW_INCLUDE_DIR})
    add_executable(arrow_flight_cpp main.cpp)
    target_link_libraries(arrow_flight_cpp PRIVATE Arrow::arrow_shared)
    target_link_libraries(arrow_flight_cpp PRIVATE ArrowFlight::arrow_flight_shared)
    target_link_libraries(arrow_flight_cpp PRIVATE ArrowFlightSql::arrow_flight_sql_shared)
  • C++ Arrow库的用法

    arrow的函数大多数是返回 arrow::Result<T> 类型,因此需要把代码写在返回值为 arrow::Result<T> 的类型的函数中,如下:

     arrow::Result <std::unique_ptr<FlightClient>> get_location() {
    ARROW_ASSIGN_OR_RAISE(auto location, Location::ForGrpcTcp("localhost", 8904));
    ARROW_ASSIGN_OR_RAISE(auto client, FlightClient::Connect(location))
    }

    ARROW_ASSIGN_OR_RAISE宏的效果是,先对右边返回值为 arrow::Result<T> 类型的表达式求值,如果出现异常,则提前return,赋上相应的Status值。

    为了方便展示,我们把代码写在lambda函数中。

    int main() {
    auto fun = []() {
    // code
    }
    fun();
    return 0;
    }
  • 验证身份获取令牌,并创建一个FlightSqlClient

    ARROW_ASSIGN_OR_RAISE(auto location, Location::ForGrpcTcp("localhost", 8904))
    ARROW_ASSIGN_OR_RAISE(auto client, FlightClient::Connect(location))
    auto user = "root";
    auto password = "";
    //Base64加密认证
    auto auth = client->AuthenticateBasicToken({}, user, password);
    ARROW_RETURN_NOT_OK(auth); // 如果result出现异常,直接return
    FlightCallOptions call_options;
    call_options.headers.push_back(auth.ValueOrDie()); //把认证放到调用选项中
    auto sql_client = std::make_unique<FlightSqlClient>(std::move(client));
  • 执行sql取得FlightInfo

    ARROW_ASSIGN_OR_RAISE(auto info, sql_client->Execute(call_options, "select now();"));
    const auto endpoints = info->endpoints();
  • 通过FlightEndPoint取回数据

    for (auto i = 0; i < endpoints.size(); i++) {
    auto &ticket = endpoints[i].ticket;
    // stream中包含数据
    ARROW_ASSIGN_OR_RAISE(auto stream, sql_client->DoGet(call_options, ticket));
    // 获取数据的Schema
    auto schema = stream->GetSchema();
    ARROW_RETURN_NOT_OK(schema);
    std::cout << "Schema:" << schema->get()->ToString() << std::endl;
    // 取得并打印数据
    while(true) {
    ARROW_ASSIGN_OR_RAISE(FlightStreamChunk chunk, stream->Next());
    if (chunk.data == nullptr) {
    break;
    }
    std::cout << chunk.data->ToString();
    }
    }

整体代码

#include <iostream>
#include <arrow/flight/api.h>
#include <arrow/flight/sql/api.h>
using namespace arrow::flight;
using namespace arrow::flight::sql;
using namespace arrow;

int main() {

auto fun = []() {
ARROW_ASSIGN_OR_RAISE(auto location, Location::ForGrpcTcp("localhost", 8904))
ARROW_ASSIGN_OR_RAISE(auto client, FlightClient::Connect(location))

auto user = "root";
auto password = "";
auto auth = client->AuthenticateBasicToken({}, user, password);
auto sql_client = std::make_unique<FlightSqlClient>(std::move(client));
ARROW_RETURN_NOT_OK(auth);
FlightCallOptions call_options;
call_options.headers.push_back(auth.ValueOrDie());

ARROW_ASSIGN_OR_RAISE(auto info, sql_client->Execute(call_options, "select now();"));
const auto endpoints = info->endpoints();
for (auto i = 0; i < endpoints.size(); i++) {
auto &ticket = endpoints[i].ticket;

ARROW_ASSIGN_OR_RAISE(auto stream, sql_client->DoGet(call_options, ticket));

auto schema = stream->GetSchema();
ARROW_RETURN_NOT_OK(schema);

std::cout << "Schema:" << schema->get()->ToString() << std::endl;
while(true) {
ARROW_ASSIGN_OR_RAISE(FlightStreamChunk chunk, stream->Next());
if (chunk.data == nullptr) {
break;
}
std::cout << chunk.data->ToString();
}
}
return Status::OK();
};

auto status = fun();

return 0;
}