使用Go BigQuery Storage API Client能否无需Protocol Buffers定义写入数据？

Go官方BigQuery Storage API客户端没有直接对应Java JsonStreamWriter的类，但可以通过以下两种方案实现无需手动编写Protocol Buffers（PB）定义即可插入数据：

1. 直接使用JSON写入模式 #

借助BigQuery Storage Write API的JSON格式支持，你可以直接构造JSON格式的行数据，无需提前生成PB文件或定义PB结构体：

import (
    "context"
    "encoding/json"
    "cloud.google.com/go/bigquery/storage/managedwriter"
)

func writeJSONRows(ctx context.Context, projectID, datasetID, tableID string, rows []map[string]interface{}) error {
    // 初始化写入客户端
    client, err := managedwriter.NewClient(ctx, projectID)
    if err != nil {
        return err
    }
    defer client.Close()

    // 构造表引用并获取表schema
    tableRef := managedwriter.TableParentFromParts(projectID, datasetID, tableID)
    schema, err := client.GetTableSchema(ctx, tableRef)
    if err != nil {
        return err
    }

    // 创建JSON格式的写入流
    stream, err := client.NewManagedStream(ctx,
        managedwriter.WithDestinationTable(tableRef),
        managedwriter.WithSchemaDescriptor(schema),
        managedwriter.WithFormat(managedwriter.JSON),
    )
    if err != nil {
        return err
    }
    defer stream.Close()

    // 序列化每行数据为JSON并写入
    for _, row := range rows {
        jsonBytes, err := json.Marshal(row)
        if err != nil {
            return err
        }
        _, err = stream.AppendRows(ctx, managedwriter.NewRowData(jsonBytes))
        if err != nil {
            return err
        }
    }

    // 完成写入并确认
    _, err = stream.Finalize(ctx)
    return err
}

这种方式依赖目标表的现有schema，只需要确保JSON字段名、类型与表schema严格匹配即可，完全不需要手动维护PB定义。

2. 动态生成PB消息 #

如果需要更灵活的类型控制，可以通过BigQuery表schema动态生成PB描述符，在运行时构造PB消息，避免手动编写.proto文件：

import (
    "context"
    "cloud.google.com/go/bigquery/storage/managedwriter"
    "google.golang.org/protobuf/types/dynamicpb"
)

func writeDynamicPB(ctx context.Context, projectID, datasetID, tableID string, rows []map[string]interface{}) error {
    client, err := managedwriter.NewClient(ctx, projectID)
    if err != nil {
        return err
    }
    defer client.Close()

    tableRef := managedwriter.TableParentFromParts(projectID, datasetID, tableID)
    schema, err := client.GetTableSchema(ctx, tableRef)
    if err != nil {
        return err
    }

    // 获取表对应的PB消息描述符
    msgDesc := schema.GetProtoDescriptor()
    stream, err := client.NewManagedStream(ctx,
        managedwriter.WithDestinationTable(tableRef),
        managedwriter.WithSchemaDescriptor(schema),
    )
    if err != nil {
        return err
    }
    defer stream.Close()

    // 动态构造PB消息并写入
    for _, row := range rows {
        msg := dynamicpb.NewMessage(msgDesc)
        for fieldName, value := range row {
            err := msg.Set(msgDesc.FindFieldByName(fieldName), value)
            if err != nil {
                return err
            }
        }
        _, err = stream.AppendRows(ctx, managedwriter.NewRowDataFromProto(msg))
        if err != nil {
            return err
        }
    }

    _, err = stream.Finalize(ctx)
    return err
}

这种方案本质还是使用PB序列化，但所有PB结构都基于表schema动态生成，无需手动编写任何PB相关代码。

注意事项 #

• 两种方案都要求数据字段与表schema严格匹配，类型不兼容或字段缺失会导致写入失败。
• 批量写入多行数据可以显著提升性能，建议合并批量请求后再发送。

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