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

Question

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

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

感谢您的帮助。

Answer 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()