如何用Go搭建本地Apache Arrow Flight服务器并在JS中获取数据

问题背景 #

我以Go作为后端服务器、JavaScript作为前端开发，此前已通过HTTP+IPC方式完成简单数据的托管与前端表格生成，但在尝试改用Apache Arrow Flight实现数据服务与获取时遇到了阻碍。核心需求是：

• 在Go后端搭建本地Apache Arrow Flight服务器，以arrow.Record格式提供数据
• 在JavaScript前端中连接该服务器并获取数据

已实现Go端的数据构建逻辑，需要补充Flight服务器的启动与数据响应逻辑；前端已有基础框架，需完善数据获取代码。

Go后端完整实现 #

要构建Flight服务器，需实现flight.Server接口的DoGet方法，用于响应客户端的数据请求。以下是完整的main.go代码：

import (
    "context"
    "fmt"
    "github.com/apache/arrow/go/v16/arrow"
    "github.com/apache/arrow/go/v16/arrow/array"
    "github.com/apache/arrow/go/v16/arrow/flight"
    "github.com/apache/arrow/go/v16/arrow/memory"
    "log"
    "net"
    "google.golang.org/grpc"
)

var metadata = arrow.NewMetadata(
    []string{"type", "round", "date"},
    []string{"int", "1dp", "2024/03/30"},
)

var schema = arrow.NewSchema([]arrow.Field{
    {Name: "X", Type: arrow.PrimitiveTypes.Int32},
    {Name: "X + 5", Type: arrow.PrimitiveTypes.Int32},
}, &metadata)

func GetPutData() arrow.Record {
    pool := memory.NewGoAllocator()
    recordBuilder := array.NewRecordBuilder(pool, schema)
    
    recordBuilder.Field(0).(*array.Int32Builder).AppendValues([]int32{1, 2, 3, 4, 5}, nil)
    recordBuilder.Field(1).(*array.Int32Builder).AppendValues([]int32{6, 7, 8, 9, 10}, nil)
    
    record := recordBuilder.NewRecord()
    return record
}

// 自定义Flight服务器，实现DoGet方法
type flightServer struct {
    flight.UnimplementedFlightServiceServer
    data arrow.Record
}

func (s *flightServer) DoGet(ctx context.Context, req *flight.Ticket) (*flight.FlightDataStream, error) {
    // 返回预定义的record数据
    return flight.NewRecordStream(s.data), nil
}

func main() {
    rec := GetPutData()
    defer rec.Release() // 确保资源释放

    // 创建gRPC服务器
    grpcServer := grpc.NewServer()
    // 注册Flight服务
    flight.RegisterFlightServiceServer(grpcServer, &flightServer{data: rec})

    // 监听本地端口
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    fmt.Println("Flight server running on localhost:50051")
    if err := grpcServer.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

关键说明 #

• 自定义flightServer结构体嵌入flight.UnimplementedFlightServiceServer，避免实现所有接口方法
• DoGet方法是核心，通过flight.NewRecordStream将arrow.Record转换为Flight数据流返回
• 启动gRPC服务器并注册Flight服务，监听50051端口（Flight默认端口）

JavaScript前端完整实现 #

使用apache-arrow库的Flight客户端连接服务器，获取数据并转换为Arrow Table。确保已安装依赖：npm install apache-arrow

import * as Arrow from 'apache-arrow';
import * as Flight from 'apache-arrow/flight';

async function runExample() {
    // 连接本地Flight服务器
    const client = Flight.connect('grpc://localhost:50051');
    
    // 请求数据（Ticket为空即可，因为我们的服务器不处理Ticket内容）
    const stream = await client.doGet(new Uint8Array());
    
    // 将数据流转换为Arrow Table
    const table = await Arrow.Table.from(stream);
    
    // 输出数据与结构
    console.table(table.toArray());
    console.log('Schema:', table.schema);
    console.log('Data:', table.data);
    
    // 关闭客户端连接
    await client.close();
}

runExample();

关键说明 #

• 使用Flight.connect建立与服务器的连接，地址格式为grpc://host:port
• doGet方法传入空Ticket（因为服务器未对Ticket做校验），获取数据流
• 通过Arrow.Table.from将数据流转换为可操作的Arrow Table

注意事项 #

• 版本一致性：确保Go端Arrow版本（示例中为v16）与前端apache-arrow版本完全一致，避免序列化/反序列化错误
• 资源释放：Go端使用defer rec.Release()确保Arrow Record资源被正确回收
• 跨域问题：如果前端在浏览器环境运行，需确保服务器配置了CORS（或使用本地开发服务器代理）

内容的提问来源于stack exchange，提问作者TheFor