Google DataFlow：在流管道中的BigQuery中插入+更新

Question

分析输入后，有两个选项可用：

• 如果x=1->插入
• 如果x=2->更新

测试

null

     UPDATE or DELETE statement over table table would affect rows in the streaming buffer, which is not supported

def insertCanonicalBQ(input):
    from google.cloud import bigquery
    client = bigquery.Client(project='project')
    dataset = client.dataset('dataset')
    table = dataset.table('table' )
    table.reload()
    table.insert_data(
        rows=[[values]])

def UpdateBQ(input):
    from google.cloud import bigquery
    import uuid
    import time
    client = bigquery.Client()
    STD= "#standardSQL"
    QUERY= STD + "\n" + """UPDATE table SET field1 = 'XXX' WHERE field2=  'YYY'"""
    client.use_legacy_sql = False    
    query_job = client.run_async_query(query=QUERY, job_name='temp-query-job_{}'.format(uuid.uuid4()))  # API request
    query_job.begin()
    while True:
         query_job.reload()  # Refreshes the state via a GET request.
         if query_job.state == 'DONE':
             if query_job.error_result:
                 raise RuntimeError(query_job.errors)
             print "done"
             return input
             time.sleep(1)

Answer 1

即使该行不在流缓冲区中，这仍然不是BigQuery中解决这个问题的方法。BigQuery存储更适合于批量突变，而不是像这样通过update突变单个实体。您的模式与我期望的事务性用例而不是分析性用例一致。

为此考虑一个基于追加的模式。每次处理实体消息时，通过流式插入将其写入BigQuery。然后，在需要时，您可以通过查询获得所有实体的最新版本。

例如，我们假设一个任意模式:IDField是您唯一的实体键/标识符，message_time表示消息发出的时间。您的实体可能有许多其他字段。要获取实体的最新版本，我们可以运行以下查询（并可能将其写入另一个表）：

#standardSQL
SELECT
  idfield,
  ARRAY_AGG(
    t ORDER BY message_time DESC LIMIT 1
  )[OFFSET(0)].* EXCEPT (idfield)
FROM `myproject.mydata.mytable` AS t
GROUP BY idfield