Как мне создать и добавить данные из CSV-файла в большой запрос и разделить таблицу с помощью Python? - PullRequest
Новая
       23

Как мне создать и добавить данные из CSV-файла в большой запрос и разделить таблицу с помощью Python?

0 голосов
/ 24 августа 2018

Я сжал файлы CSV GZIP в облачном хранилище Google и, используя Python, автоматически определяю схему и создаю новую таблицу в Google BigQuery в зависимости от соглашения об именах.Как разделить создаваемую таблицу?У меня уже есть столбец Дата в данных, которые я хотел бы использовать.

# importing libraries
from google.cloud import bigquery

# defining first load list
first_load_list = []

#defining tracker file
tracker_file = open("tracker_file", "a")

#reading values from config file
config_file = open("ingestion.config", "r")
for line in config_file:
    if "project_id" in line:
        project_id = line.split("=")[1].strip()
    elif "dataset" in line:
        dataset = line.split("=")[1].strip()
    elif "gcs_location" in line:
        gcs_location = line.split("=")[1].strip()
    elif "bq1_target_table" in line:
        bq1_target_table = line.split("=")[1].strip()
    elif "bq2_target_table" in line:
        bq2_target_table = line.split("=")[1].strip()
    elif "bq1_first_load_filename" in line:
        bq1_first_load_filename = line.split("=")[1].strip()
        first_load_list.append(bq1_first_load_filename)
    elif "bq2_first_load_filename" in line:
        bq2_first_load_filename = line.split("=")[1].strip()
        first_load_list.append(bq2_first_load_filename)
    elif "gcs_bucket" in line:
        gcs_bucket = line.split("=")[1].strip()

# reading bucket list temp file
bucket_list_file = open("bucket_list.temp", "r")
bucket_list = []
for entry in bucket_list_file:
    bucket_list.append(entry)


# defining client and specifying project
client = bigquery.Client(project_id)
dataset_id = dataset
dataset_ref = client.dataset(dataset_id)
job_config = bigquery.LoadJobConfig()
job_config.autodetect = True
job_config.skip_leading_rows = 1
job_config.source_format = bigquery.SourceFormat.CSV


# leading files into tables based on naming convention
for filename in first_load_list:
    if "BQ2_2" in filename:
        uri = gcs_location + filename
        print "Processing file = " + uri
        load_job = client.load_table_from_uri(
            uri.strip(),
            dataset_ref.table(bq2_target_table),
            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'
        assert client.get_table(dataset_ref.table(bq2_target_table))
        tracker_file.write(filename + "\n")
        print filename.strip() + " processing complete\n"
    elif "BQ1_2" in filename:
        uri = gcs_location + filename
        print "Processing file = " + uri
        load_job = client.load_table_from_uri(
            uri.strip(),
            dataset_ref.table(bq1_target_table),
            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'
        assert client.get_table(dataset_ref.table(bq1_target_table))
        tracker_file.write(filename + "\n")
        print filename.strip() + " processing complete\n"

tracker_file.close()

Это код, который я запускаю для первой загрузки.Как только первые таблицы загрузки созданы, я только хочу добавить данные в эти таблицы в будущем.Я посмотрел на https://cloud.google.com/bigquery/docs/creating-partitioned-tables, но я не могу понять, как реализовать в Python.

Может кто-нибудь помочь, пожалуйста, указать мне правильное направление?

1 Ответ

0 голосов
/ 25 августа 2018

Вы можете использовать job_config._properties['load']['timePartitioning'] = {"type":"DAY", 'field':'your_field'} для создания таблицы разделов при загрузке.Я только что проверил его с помощью тестовых данных, и он работал как положено.

Обратите внимание, что раздел с API пока поддерживает только 'DAY'.

См. Проблему с GitHub

...