Google GCP Cloud Functions 到 BigQuery 错误答案

【问题标题】：Google GCP Cloud Functions to BigQuery ErrorGoogle GCP Cloud Functions 到 BigQuery 错误
【发布时间】：2020-07-23 01:51:44
【问题描述】：

我创建了一个用于将数据发送到 BigQuery 的 Cloud Functions Cloud Functions 正在从 pub/sub 接收数据。

场景 1：我写了一个python代码直接发送JSON数据到Bigquery，没问题

场景 2：我将 JSON 数据保存到 .json 文件，并使用 bq load 命令手动上传到 Bigquery，没问题

场景 3：（出现错误的地方） Cloud Functions 可以从 Pub/Sub 接收数据，但无法将其发送到 BigQuery。

这是 Cloud Functions 的代码：

from google.cloud import bigquery
import base64, json, sys, os

def pubsub_to_bq(event, context):
   if 'data' in event:
      print("Event Data is found : " + str(event['data']))
      name = base64.b64decode(event['data']).decode('utf-8')
   else:
      name = 'World'
   print('Hello {}!'.format(name))


   pubsub_message = base64.b64decode(event['data']).decode('utf-8')
   print(pubsub_message)
   to_bigquery(os.environ['dataset'], os.environ['table'], json.loads(str(pubsub_message)))

def to_bigquery(dataset, table, document):
   bigquery_client = bigquery.Client()
   table = bigquery_client.dataset(dataset).table(table)
   
   job_config.source_format = bq.SourceFormat.NEWLINE_DELIMITED_JSON
   job_config = bq.LoadJobConfig()
   job_config.autodetect = True
   
   errors = bigquery_client.insert_rows_json(table,json_rows=[document],job_config=job_config)
   if errors != [] :
      print(errors, file=sys.stderr)

我已经尝试了两种类型的 JSON 数据格式，但没有运气。 [{"field1":"data1","field2":"data2"}] 或 {"field1":"data1","field2":"data2"}

我可以从 Cloud Functions 事件日志中获得的所有错误消息是： textPayload: "函数执行耗时 100 毫秒，完成状态为：'crash'"

任何专家可以帮助我吗？谢谢。

【问题讨论】：

为什么要这样做？而是编写一个数据流作业。并且有可用的数据流模板
您的错误可能与正在调查的已知问题here 相关，并且正在修复中。

标签： python google-cloud-platform google-bigquery google-cloud-functions

【解决方案1】：

如果你有look to the library code，insert_rows_json 你有这个

    def insert_rows_json(
        self,
        table,
        json_rows,
        row_ids=None,
        skip_invalid_rows=None,
        ignore_unknown_values=None,
        template_suffix=None,
        retry=DEFAULT_RETRY,
        timeout=None,
    ):

没有job_config 参数！崩溃应该是这个错误造成的

insert_rows_json performs a streaming insert 方法而不是加载作业。

对于来自 JSON 的加载作业，您可以使用 load_table_from_json 方法，您也可以在库的源代码中找到该方法。代码库是similar to this (for the JobConfig option)

    def load_table_from_json(
        self,
        json_rows,
        destination,
        num_retries=_DEFAULT_NUM_RETRIES,
        job_id=None,
        job_id_prefix=None,
        location=None,
        project=None,
        job_config=None,
    ):

【讨论】：