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

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

安装

导入bqwriter包：

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

安装命令：

go get github.com/OTA-Insight/bqwriter

基本示例

基本InsertAll Streamer

import (
    "context"

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

// 创建上下文
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()

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

定义行数据结构：

import (
    "cloud.google.com/go/bigquery"
    "cloud.google.com/go/civil"
)

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(rr.Timestamp),
        "username":  mr.Username,
    }, "", nil
}

自定义InsertAll Streamer

import (
    "context"

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

ctx := context.Background()

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

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

Storage Streamer

import (
    "context"

    "github.com/OTA-Insight/bqwriter"
    "google.golang.org/protobuf/reflect/protodesc"

    // 定义实际预编译protobuf Go代码的路径
    "path/to/my/proto/package/protodata"
)

ctx := context.Background()

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

// 创建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)

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

Batch Streamer

import (
    "context"
    "path/filepath"

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

func main() {
    ctx := context.Background()
	
    // 使用默认配置创建BatchClientConfig
    batchConfig := new(bqwriter.BatchClientConfig)
	
    // 创建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("/usr/joe/my/data/path/exported_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 Application Default Credentials进行授权。这允许您的应用程序在许多环境中运行而无需显式配置。

性能考虑

• 对于高性能场景，建议使用Storage API而非InsertAll API
• 适当设置WorkerCount以匹配您的需求
• 调整MaxBatchDelay和BatchSize以获得最佳性能
• 使用ProtobufDescriptor而非BigQuerySchema以获得更好的性能

常见问题

InsertAll Streamer似乎每请求只插入一行而不是批量插入

确保您的配置与您的带宽需求匹配。不要使用超过您需要的工作线程数（WorkerCount），并确保MaxBatchDelay和BatchSize值配置适当。