使用 BigQuery Write API 在 Golang 中编写数据

Golang不知道大家是否熟悉？今天我将给大家介绍《使用 BigQuery Write API 在 Golang 中编写数据》，这篇文章主要会讲到等等知识点，如果你在看完本篇文章后，有更好的建议或者发现哪里有问题，希望大家都能积极评论指出，谢谢！希望我们能一起加油进步！

我正在尝试使用新的 bigquery storage api 从 golang 进行流式插入。根据此页面，我了解到该 api 取代了旧的流式插入 bigquery api。

但是，文档中的任何示例都没有显示如何实际插入行。为了创建 appendrowsrequest，我得到了以下结果：

&storagepb.AppendRowsRequest{
    WriteStream: resp.Name,
    Rows: &storagepb.AppendRowsRequest_ProtoRows{
        ProtoRows: &storagepb.AppendRowsRequest_ProtoData{
            WriterSchema: nil, // protobuf schema??
            Rows: &storagepb.ProtoRows{
                SerializedRows: [][]byte{}, // serialized protocol buffer data??
            },
        },
    },
}

我应该将哪些数据放入上面的 serializedrows 字段中？

上面的 storagepb.protorows 结构记录在此处。不幸的是，给出的只是一个指向协议缓冲区主要概述页面的链接。

谁能给我一个使用新的 bigquery storage api 将行从 golang 流式传输到 bigquery 的示例吗？

正确答案

在上述答案的帮助下，我得到了一个工作示例，可以在 github 上找到： https://github.com/alexflint/bigquery-storage-api-example

主要代码为：

const (
    project = "myproject"
    dataset = "mydataset"
    table   = "mytable"
    trace   = "bigquery-writeclient-example" // identifies this client for bigquery debugging
)

// the data we will stream to bigquery
var rows = []*row{
    {name: "john doe", age: 104},
    {name: "jane doe", age: 69},
    {name: "adam smith", age: 33},
}

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

    // create the bigquery client
    client, err := storage.newbigquerywriteclient(ctx)
    if err != nil {
        log.fatal(err)
    }
    defer client.close()

    // create the write stream
    // a committed write stream inserts data immediately into bigquery
    resp, err := client.createwritestream(ctx, &storagepb.createwritestreamrequest{
        parent: fmt.sprintf("projects/%s/datasets/%s/tables/%s", project, dataset, table),
        writestream: &storagepb.writestream{
            type: storagepb.writestream_committed,
        },
    })
    if err != nil {
        log.fatal("createwritestream: ", err)
    }

    // get the stream by calling appendrows
    stream, err := client.appendrows(ctx)
    if err != nil {
        log.fatal("appendrows: ", err)
    }

    // get the protobuf descriptor for our row type
    var row row
    descriptor, err := adapt.normalizedescriptor(row.protoreflect().descriptor())
    if err != nil {
        log.fatal("normalizedescriptor: ", err)
    }

    // serialize the rows
    var opts proto.marshaloptions
    var data [][]byte
    for _, row := range rows {
        buf, err := opts.marshal(row)
        if err != nil {
            log.fatal("protobuf.marshal: ", err)
        }
        data = append(data, buf)
    }

    // send the rows to bigquery
    err = stream.send(&storagepb.appendrowsrequest{
        writestream: resp.name,
        traceid:     trace, // identifies this client
        rows: &storagepb.appendrowsrequest_protorows{
            protorows: &storagepb.appendrowsrequest_protodata{
                // protocol buffer schema
                writerschema: &storagepb.protoschema{
                    protodescriptor: descriptor,
                },
                // protocol buffer data
                rows: &storagepb.protorows{
                    serializedrows: data, // serialized protocol buffer data
                },
            },
        },
    })
    if err != nil {
        log.fatal("appendrows.send: ", err)
    }

    // get the response, which will tell us whether it worked
    _, err = stream.recv()
    if err != nil {
        log.fatal("appendrows.recv: ", err)
    }

    log.println("done")
}

上面“row”结构的协议缓冲区定义是：

syntax = "proto3";

package tutorial;

option go_package = ".;main";

message row {
    string name = 1;
    int32 age = 2;
}

您需要首先使用与协议缓冲区对应的架构创建一个bigquery数据集和表。请参阅上面链接的存储库中的自述文件，了解如何执行此操作。

运行上面的代码后，数据显示在bigquery中，如下所示：

$ bq query 'select * from mydataset.mytable'
Waiting on bqjob_r1b39442e5474a885_0000017df21f629e_1 ... (0s) Current status: DONE   
+------------+-----+
|    name    | age |
+------------+-----+
| John Doe   | 104 |
| Jane Doe   |  69 |
| Adam Smith |  33 |
+------------+-----+

感谢大家的帮助！

我找到了一些有关将流写入表的文档 [1][2]，但我不太确定这就是您正在寻找的内容。请记住，storage/apiv1beta2 目前处于测试状态，因此可能尚未实现或缺乏相关文档。如果我附加的文档对您没有帮助，我们可以打开一个公共问题跟踪器来正确记录或实现行流。

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~