将 JSON 从 googlecloud 存储上传到 bigquery 时出现混淆

Posted 2023-03-24

【中文标题】将 JSON 从 googlecloud 存储上传到 bigquery 时出现混淆

【英文标题】：Confusion when uploading a JSON from googlecloud storage to bigquery

【发布时间】：2017-07-03 01:38:12

【问题描述】：

你好，这是一个两部分的问题

1) 目前我正在尝试通过 python 脚本将文件从谷歌云存储上传到 bigquery。我正在尝试按照谷歌帮助网站给出的步骤进行操作。

https://cloud.google.com/bigquery/docs/loading-data-cloud-storage#bigquery-import-gcs-file-python

def load_data_from_gcs(dataset_name, table_name, source):
    bigquery_client = bigquery.Client()
    dataset = bigquery_client.dataset(dataset_name)
    table = dataset.table(table_name)
    job_name = str(uuid.uuid4())

    job = bigquery_client.load_table_from_storage(
        job_name, table, source)

    job.begin()

    wait_for_job(job)

    print('Loaded  rows into :.'.format(
        job.output_rows, dataset_name, table_name))

我不确定要为“load_data_from_gcs”的第一行添加什么，因为在谷歌云中没有表格，它是我要上传的 JSON 文件。 “表”部分是我要创建的表的名称还是在谈论存储桶，因为没有部分可以指定我要从哪个存储桶中提取。

这是我目前的代码。

import json
import argparse
import time
import uuid

from google.cloud import bigquery
# from google.cloud import storage

def load_data_from_gcs('dataworks-356fa', table_name, 'pullnupload.json'):
    bigquery_client = bigquery.Client('dataworks-356fa')
    dataset = bigquery_client.dataset('FirebaseArchive')
    table = dataset.table(table_name)
    job_name = str(uuid.uuid4())

    job = bigquery_client.load_table_from_storage(
        job_name, table, source)

    job.begin()

    wait_for_job(job)

    print('Loaded  rows into :.'.format(
        job.output_rows, dataset_name, table_name))

第 2 部分） 我希望这个脚本每周运行一次，并且能够删除旧表并创建一个新表，或者只过滤非重复数据。哪个更容易。

感谢您的帮助。

【参考方案1】：

不确定您遇到了什么问题，但将数据从 GCS 文件加载到 BigQuery 正是您已经在做的事情。

如果您有一个具有此架构的表：

["name": "id", "type": "INT64", "name": "name", "type": "STRING"]

如果您在 GCS 中有此文件（例如位于“gs://bucket/json_data.json”）：

"id": 1, "name": "test1"
"id": 2, "name": "test2"

您现在只需要设置 job 对象来处理 JSON 文件作为输入，如下所示：

def load_data_from_gcs('dataworks-356fa', table_name, 'pullnupload.json'):
    bigquery_client = bigquery.Client('dataworks-356fa')
    dataset = bigquery_client.dataset('FirebaseArchive')
    table = dataset.table(table_name)
    job_name = str(uuid.uuid4())

    job = bigquery_client.load_table_from_storage(
        job_name, table, "gs://bucket/json_data.json")

    job.source_format = 'NEWLINE_DELIMITED_JSON'
    job.begin()

就是这样。

（如果你有一个 CSV 文件，那么你必须相应地设置你的 job 对象）。

至于第二个问题，实际上是尝试不同的方法，看看哪种方法最适合你。

要删除一个表，你只需要运行：

table.delete()

要从表中删除重复数据，一种可能性是编写删除重复数据并将结果保存到同一个表的查询。比如：

query_job = bigquery_client.run_async_query(query=your_query, job_name=job_name)
query_job.destination = Table object
query_job.write_disposition = 'WRITE_TRUNCATE'
query_job.begin()

【讨论】：

所以当我在 python 中运行它时，它将运行没有错误，但它不执行任何操作。有什么建议吗？

你的意思是它不处理任何东西？ job.state 继续 RUNNING?

不，它让我回到正常终端，就好像工作已完成并且没有显示任何错误，但我的谷歌云或 bigquery 没有任何变化

您是否正在等待job 达到DONE 状态以查看错误消息？