栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 面试经验 > 面试问答

从代码创建Golang代码后,它并未插入BigQuery的表中

面试问答 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

从代码创建Golang代码后,它并未插入BigQuery的表中

根据旧答案,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 insertuploader := 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 {}    // delete the previous table    if err := client.Dataset(dataset).Table(table).Delete(context.Background()); err != nil {}    // create the new table    schema, err := bigquery.InferSchema(row{})    if err != nil {}    if err := client.Dataset(dataset).Table(table).Create(context.Background(), &bigquery.Tablemetadata{Schema: schema}); err != nil {}    // 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 {}         }(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};returnb}

这是我完整的工作示例。



转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/413155.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号