Возобновление загрузки в BigQuery с использованием интерфейса Pandas

Question

Я использую BigQuery Python API вместе с соединителем BigQuery для Pandas .

Каждый раз, когда я append обращаюсь к набору данных в BigQuery, я хочуубедитесь, что я начинаю с того места, где ушел в прошлый раз, чтобы предотвратить дублирование и потерю данных.

Есть ли в Загрузить конфигурацию задания параметр или что-то еще, что будет делать это автоматически?Если нет, как вы предлагаете мне обрабатывать ошибки подключения и загружать файлы, не допуская дублирования данных?Я знаю, что могу запросить последнюю строку и добавить данные, основанные на этом, но я предпочитаю не запрашивать, так как BigQuery взимает плату за запросы.

Вот что у меня есть для загрузки в BigQuery:

import pandas as pd
from google.cloud import bigquery, exceptions

test_df = pd.DataFrame({
    'num_legs': [2, 4, 8, 0],
    'num_wings': [2, 0, 0, 0],
    'num_specimen_seen': [10, 2, 1, 8],
    'names': ['falcon', 'dog', 'spider', 'fish']
})
project = "test-project"
dataset_id = "test-dataset"
table_id = "test-table"
client = bigquery.Client()
try:
    dataset_ref = client.dataset(dataset_id=dataset_id, project=project)
    dataset = client.get_dataset(dataset_ref)
except exceptions.NotFound:
    print("specified dataset not found! -- creating a new dataset.")
    dataset = client.create_dataset(dataset_id)

table_ref = dataset.table(table_id)
load_job = client.load_table_from_dataframe(
    dataframe=test_df, destination=table_ref, project=project)
load_job.result()

Answer 1

При условии, что вы можете сгенерировать уникальный идентификатор загрузки-пакета поверх исходных данных.

Один из способов достижения вашей цели:

1. Пометьте свою загрузочную работу с помощью load_batch_id

   Client.load_table_from_dataframe имеет аргумент: job_config (google.cloud.bigquery.job. LoadJobConfig , необязательно)

   Вставить load_batch_id в LoadJobConfig.labels (Dict [str, str] - метки для задания.)

2. Если вам необходимо подтвердить, успешно ли загружена работа, используйте job.list api для поиска работы с меткой load_batch_id.

В одном упрощенном случае, скажем, вы загружаете данные только ежедневно в BQ, а ваши исходные данные группируются по дате. Затем ваш сценарий Python 20190325 начинается с первой проверки задания с меткой 20190324 (или глубже в прошлое), чтобы выяснить, не нужно ли повторить попытку.