Сохранить результаты Bigquery в JSON в Google Composer - PullRequest
Новая
       27

Сохранить результаты Bigquery в JSON в Google Composer

0 голосов
/ 18 марта 2020

Я создал ниже DAG для ежедневного запуска сценария sql. Как сохранить результаты запроса в файл JSON и сохранить в папке DAG в Google Composer? 

import datetime
import airflow
from airflow.operators import bash_operator
from airflow.contrib.operators import bigquery_operator

START_DATE = datetime.datetime(2020, 3, 1)

default_args = {
    'owner': 'Alen',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=15),
    'start_date': START_DATE,
}

with airflow.DAG(
        'Dag_Name',
        'catchup=False',
        default_args=default_args,
        schedule_interval=datetime.timedelta(days=1)) as dag:

    task_name = bigquery_operator.BigQueryOperator(
        task_id='task_name',
        sql= 'query.sql',
        use_legacy_sql=False,
        write_disposition= 'WRITE_TRUNCATE' ,        
        destination_dataset_table='Project.Dataset.destination_table')

1 Ответ

2 голосов
/ 18 марта 2020

Одной из альтернатив является запуск экспорта из BQ в GCS с папкой DAG в качестве места назначения.

Вы можете использовать оператор bash или bq

Затем запустите что-то вроде этого в конце вашего скрипта: 

copy_files_to_DAG_folder = bash_operator.BashOperator(
    task_id='Copy_files_to_GCS',
    bash_command='bq extract --destination_format JSON--print_header=false 'BQ_TABLE' 
    'GCS_DAG_FOLDER_LOCATION''

Из документов: 

 bq --location=location extract \
 --destination_format format \
 --compression compression_type \
 --field_delimiter delimiter \
 --print_header=boolean \
 project_id:dataset.table \
 gs://bucket/filename.ext
...