使用 Python 将数据从 datasteam 加载到 BigQuery答案

【问题标题】：Loading data from datasteam to BigQuery using Python使用 Python 将数据从 datasteam 加载到 BigQuery
【发布时间】：2018-12-04 21:53:51
【问题描述】：

我正在使用 API 调用从服务中检索数据。数据是嵌套的 Json，其中包含可能还包含 Json 对象的数组。

示例：

基本上我想将它上传到 BigQuery 中的表中。我为每个数组创建了一个表。 Jason 对象在同一张表中解压。例如：

Orders: All customer fields, all ShippingAdress, orderDateUtc etc..
Orders_items:  orderid, discountEach, giftTo etc..
Order_items_historicalCategories: ....

我不确定最好的方法。我可以从 API 调用（数据流）创建 CSV 文件，然后使用 COPY per CSV 上传它们，但这似乎过多。我正在寻找一种跳过 CSV 创建的方法。

是否有可以处理这些数据并将其直接上传到表的操作员或包？我假设许多其他组织已经完成了我需要做的事情，但我没有在文档中看到任何“内置”方法来做到这一点 https://cloud.google.com/bigquery/docs/loading-data

任何帮助将不胜感激。

【问题讨论】：

标签： python google-bigquery

【解决方案1】：

您基本上必须遵循有关如何使用load json data using python 的文档，使用nested and repeated fields。例如，使用后一个链接中的模式，您可以通过以下方式加载嵌套和重复的 JSON 数据（您可以使用找到的示例数据进行测试 here）：

import sys

def load_nested_json():
    from google.cloud import bigquery
    client = bigquery.Client()

    dataset_id, table_id, uri = sys.argv[1:]
    dataset_ref = client.dataset(dataset_id)
    job_config = bigquery.LoadJobConfig()
    job_config.schema = [
        bigquery.SchemaField('id', 'STRING', mode='NULLABLE'),
        bigquery.SchemaField('first_name', 'STRING', mode='NULLABLE'),
        bigquery.SchemaField('last_name', 'STRING', mode='NULLABLE'),
        bigquery.SchemaField('dob', 'DATE', mode='NULLABLE'),
        bigquery.SchemaField('addresses', 'RECORD', mode='REPEATED', fields=[
            bigquery.SchemaField('status', 'STRING', mode='NULLABLE'),
            bigquery.SchemaField('address', 'STRING', mode='NULLABLE'),
            bigquery.SchemaField('city', 'STRING', mode='NULLABLE'),
            bigquery.SchemaField('state', 'STRING', mode='NULLABLE'),
            bigquery.SchemaField('zip', 'STRING', mode='NULLABLE'),
            bigquery.SchemaField('numberOfYears', 'STRING', mode='NULLABLE'),
        ]),
    ]
    table_ref = dataset_ref.table(table_id)
    # Uncomment following lines to also create the destination table
    # table = bigquery.Table(table_ref, job_config.schema)
    # table = client.create_table(table)

    # print('Created table {}'.format(table.full_table_id))

    job_config.source_format = "NEWLINE_DELIMITED_JSON"

    load_job = client.load_table_from_uri(
        uri,
        table_ref,
        job_config=job_config)  # API request

    assert load_job.job_type == 'load'

    load_job.result()  # Waits for table load to complete.

    assert load_job.state == 'DONE'

if __name__ == '__main__':
    load_nested_json()

【讨论】：

谢谢。为什么 create_table 调用？表不应该是预先创建的吗？另外，我必须硬编码字段名称吗？假设我有 16 个不同的 API 调用，它不能获取 Json 文件的列名吗？
您没有拥有来创建表格。我改变了我的答案以反映这一点。 python 客户端库不支持从 JSON 文件加载模式（检查 here），但这是已经请求的功能，目前在 "To Do" list 中。
好的...但是这种方法的问题是我必须列出代码中的所有列。这不是很通用，我只有 1 个 API 调用就有 150 列。（我还有 6 个 API 调用）。这是唯一的方法吗？
你可以编写一个方法来解析带有表模式的 json 文件并将输出传递给LoadJobConfig().schema。我认为这是你可以用 python 得到的最好的东西，除非你看到我在docs 中缺少的东西。您还可以使用bq cli，它支持将模式作为 json 文件传递（查找bq load）。