我有一个使用此架构的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连接?
根据旧答案,BigQuery流缓冲区可能需要最多2分钟的时间才能正确地附加到已删除并立即重新创建的表上。
我进行了一些测试,以我为例,该表仅用了几秒钟的时间,而不是其他问题报告的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) {}即可,直到可以安全插入行为止。
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()不喜欢int,因此必须包含func min(a,b int)int{if a<b{return a};return b}。
math.Min()
func min(a,b int)int{if a<b{return a};return b}
这是我完整的工作示例。