在我从代码创建 Golang 代码后，它没有插入到 BigQuery 的表中

Question

我有一个具有此架构的 BigQuery 表：

name    STRING  NULLABLE    
age     INTEGER NULLABLE    
amount  INTEGER NULLABLE

并且我可以使用以下代码成功插入表格：

ctx := context.Background()
client, err := bigquery.NewClient(ctx, projectID)
if err != nil {
    log.Fatal(err)
}

u := client.Dataset(dataSetID).Table("test_user").Uploader()

savers := []*bigquery.StructSaver{
    {Struct: test{Name: "Sylvanas", Age: 23, Amount: 123}},
}

if err := u.Put(ctx, savers); err != nil {
    log.Fatal(err)
}
fmt.Printf("rows inserted!!")

这很好，因为该表已经在 bigquery 上创建，我现在要做的是删除该表（如果存在）并从代码中再次创建它：

type test struct {
    Name   string
    Age    int
    Amount int
}

if err := client.Dataset(dataSetID).Table("test_user").Delete(ctx); err != nil {
    log.Fatal(err)
}

fmt.Printf("table deleted")

t := client.Dataset(dataSetID).Table("test_user")

// Infer table schema from a Go type.
schema, err := bigquery.InferSchema(test{})

if err := t.Create(ctx,
    &bigquery.TableMetadata{
        Name:           "test_user",
        Schema:         schema,
    }); err != nil {
    log.Fatal(err)
}

fmt.Printf("table created with the test schema")

这也很好用，因为它正在删除表并使用我的结构测试中推断的架构创建它。

当我在删除/创建过程之后尝试执行上述插入时，问题就来了。没有抛出错误，但它没有插入数据（如果我评论删除/创建部分，插入工作正常）。

我做错了什么？ 我是否需要以某种方式提交创建表事务才能插入，或者我是否需要关闭 DDBB 连接？

Answer 1

根据此old answer，可能需要最多 2 分钟才能将 BigQuery 流式缓冲区正确附加到已删除并立即重新创建的表。

我已经进行了一些测试，在我的情况下，只需要几秒钟就可以得到表格，而不是其他问题上报告的 2~5 分钟。生成的代码与您的完全不同，但概念应该适用。

我尝试的是，不是直接插入行，而是将它们添加到缓冲通道中，然后等到您可以验证当前表是否正确保存值后再开始发送。

我使用了一个非常简单的结构来运行我的测试（因此编写代码更容易）：

type Row struct {
    ByteField []byte
}

我通过以下方式生成我的行：

func generateRows(rows chan<- *Row) {
    for {
            randBytes := make([]byte, 100)
            _, _ = rand.Read(randBytes)
            rows <- &row{randBytes}
            time.Sleep(time.Millisecond * 500) // use whatever frequency you need to insert rows at
    }
}

注意我是如何将行发送到频道的。您只需从数据源中获取它们，而不是生成它们。

下一部分是寻找一种方法来检查表是否正确保存了行。我所做的是尝试将其中一个缓冲行插入表中，恢复该行，并验证一切是否正常。如果没有正确返回该行，则将其发送回缓冲区。

func unreadyTable(rows chan *row) bool {
    client, err := bigquery.NewClient(context.Background(), project)
    if err != nil {return true}

    r := <-rows // get a row to try to insert       
    uploader := client.Dataset(dataset).Table(table).Uploader()
    if err := uploader.Put(context.Background(), r); err != nil {rows <- r;return true}

    i, err := client.Query(fmt.Sprintf("select * from `%s.%s.%s`", project, dataset, table)).Read(context.Background())
    if err != nil {rows <- r; return true}
    var testRow []bigquery.Value
    if err := i.Next(&testRow); err != nil {rows <- r;return true}
    if reflect.DeepEqual(&row{testrow[0].([]byte)}, r) {return false} // there's probably a better way to check if it's equal
    rows <- r;return true
}

使用这样的函数，我们只需要添加for ; unreadyTable(rows); time.Sleep(time.Second) {} 即可阻塞，直到可以安全地插入行。

最后，我们把所有东西放在一起：

func main() {

    // initialize a channel where the rows will be sent
    rows := make(chan *row, 1000) // make it big enough to hold several minutes of rows

    // start generating rows to be inserted
    go generateRows(rows)

    // create the BigQuery client
    client, err := bigquery.NewClient(context.Background(), project)
    if err != nil {/* handle error */}

    // delete the previous table
    if err := client.Dataset(dataset).Table(table).Delete(context.Background()); err != nil {/* handle error */}

    // create the new table
    schema, err := bigquery.InferSchema(row{})
    if err != nil {/* handle error */}
    if err := client.Dataset(dataset).Table(table).Create(context.Background(), &bigquery.TableMetadata{Schema: schema}); err != nil {/* handle error */}

    // wait for the table to be ready
    for ; unreadyTable(rows); time.Sleep(time.Second) {}

    // once it's ready, upload indefinitely
    for {
            if len(rows) > 0 { // if there are uninserted rows, create a batch and insert them
                    uploader := client.Dataset(dataset).Table(table).Uploader()
                    insert := make([]*row, min(500, len(rows))) // create a batch of all the rows on buffer, up to 500
                    for i := range insert {insert[i] = <-rows}
                    go func(insert []*row) { // do the actual insert async
                            if err := uploader.Put(context.Background(), insert); err != nil {/* handle error */}
                    }(insert)
            } else { // if there are no rows waiting to be inserted, wait and check again
                    time.Sleep(time.Second)
            }
    }
}

注意：由于math.Min() 不喜欢整数，所以我必须包含func min(a,b int)int{if a<b{return a};return b}。

这是我的full working example。