golang高效写入Google BigQuery数据的高性能插件库bqwriter的使用

bqwriter是一个Go语言包，用于以高吞吐量并发地将数据写入Google BigQuery。它默认使用InsertAll() API（基于REST API），但也可以配置使用Storage Write API（基于GRPC）。

安装

import "github.com/OTA-Insight/bqwriter"

在你的项目目录中运行：

go get github.com/OTA-Insight/bqwriter

注意：该包仍在开发中，可能会进行不兼容的更改。

支持的Go版本

目前支持Go 1.17及更高版本。

示例

基本InsertAll Streamer

import (
    "context"
    "time"

    "github.com/OTA-Insight/bqwriter"
)

// 定义行数据结构
type myRow struct {
    Timestamp time.Time
    Username  string
}

func (mr *myRow) Save() (row map[string]bigquery.Value, insertID string, err error) {
    return map[string]bigquery.Value{
        "timestamp": civil.DateTimeOf(mr.Timestamp),
        "username":  mr.Username,
    }, "", nil
}

func main() {
    ctx := context.Background()

    // 创建BQ写入客户端
    bqWriter, err := bqwriter.NewStreamer(
        ctx,
        "my-gcloud-project",
        "my-bq-dataset",
        "my-bq-table",
        nil, // 使用默认配置
    )
    if err != nil {
        panic(err)
    }
    defer bqWriter.Close()

    // 写入数据
    bqWriter.Write(&myRow{Timestamp: time.Now(), Username: "test"})
}

自定义InsertAll Streamer

import (
    "context"
    "time"

    "github.com/OTA-Insight/bqwriter"
)

func main() {
    ctx := context.Background()

    // 创建自定义配置的BQ写入客户端
    bqWriter, err := bqwriter.NewStreamer(
        ctx,
        "my-gcloud-project",
        "my-bq-dataset",
        "my-bq-table",
        &bqwriter.StreamerConfig{
            WorkerCount: 5, // 使用5个工作线程
            InsertAllClient: &bqwriter.InsertAllClientConfig{
                FailOnInvalidRows:    true,
                FailForUnknownValues: true,
            },
        },
    )
    if err != nil {
        panic(err)
    }
    defer bqWriter.Close()

    bqWriter.Write(&myRow{Timestamp: time.Now(), Username: "test"})
}

Storage Streamer

import (
    "context"

    "github.com/OTA-Insight/bqwriter"
    "google.golang.org/protobuf/reflect/protodesc"
    "path/to/my/proto/package/protodata"
)

func main() {
    ctx := context.Background()

    // 创建proto描述符
    protoDescriptor := protodesc.ToDescriptorProto((&protodata.MyCustomProtoMessage{}).ProtoReflect().Descriptor())

    // 创建Storage API驱动的BQ写入客户端
    bqWriter, err := bqwriter.NewStreamer(
        ctx,
        "my-gcloud-project",
        "my-bq-dataset",
        "my-bq-table",
        &bqwriter.StreamerConfig{
            WorkerCount: 5,
            StorageClient: &bqwriter.StorageClientConfig{
                ProtobufDescriptor: protoDescriptor,
            },
        },
    )
    if err != nil {
        panic(err)
    }
    defer bqWriter.Close()

    // 创建并填充proto消息
    msg := new(protodata.MyCustomProtoMessage)
    // TODO: 填充msg字段

    // 写入数据
    bqWriter.Write(msg)
}

Batch Streamer

import (
    "context"
    "os"
    "path/filepath"

    "github.com/OTA-Insight/bqwriter"
    "cloud.google.com/go/bigquery"
)

func main() {
    ctx := context.Background()
    
    // 创建Batch配置
    batchConfig := new(bqwriter.BatchClientConfig)
    
    // 创建Batch驱动的BQ写入客户端
    bqWriter, err := bqwriter.NewStreamer(
        ctx,
        "my-gcloud-project",
        "my-bq-dataset",
        "my-bq-table",
        &bqwriter.StreamerConfig{
            BatchClient: batchConfig,
        },
    )
    if err != nil {
        panic(err)
    }
    defer bqWriter.Close()

    // 读取数据文件
    files, err := filepath.Glob("/path/to/data/*.json")
    if err != nil {
        panic(err)
    }
    
    for _, fp := range files {
        file, err := os.Open(fp)
        if err != nil {
            panic(err)
        }

        // 写入数据到BigQuery
        err = bqWriter.Write(file)
        if err != nil {
            panic(err)
        }
    }
}

授权

bqwriter使用Google应用默认凭证进行API端点调用的授权。这允许你的应用在许多环境中运行而无需显式配置。

错误处理

当前版本的bqwriter采用"fire-and-forget"（发送即忘记）的哲学设计。实际的写入错误发生在异步工作goroutine上并且只被记录。你可以通过实现自己的日志记录器来获取这些日志。

贡献

欢迎贡献。请参阅CONTRIBUTING文档了解详情。请注意，本项目发布时附带贡献者行为准则。参与本项目即表示你同意遵守其条款。

开发者指南

测试

使用以下命令运行测试：

go test -v ./...
golangci-lint run

你还可以运行集成测试：

go run ./internal/test/integration --help

常见问题

我的insertAll streamer似乎每请求插入1行而不是批量插入，这是怎么回事？

确保你的配置符合你的带宽需求。不要使用超过你需要的worker数量（WorkerCount）。同时确保MaxBatchDelay和BatchSize值配置得当。