Airflow не может перебрать список xcom_pull с помощью GoogleCloud Operatos - PullRequest
Новая
       6

Airflow не может перебрать список xcom_pull с помощью GoogleCloud Operatos

0 голосов
/ 04 июля 2019

Я хотел бы динамически получить список CSV-файлов из корзины gcs, а затем выгрузить каждый из них в соответствующую таблицу BQ.

Я использую GoogleCloudStorageListOperator и GoogleCloudStorageToBigQueryOperator операторов 

GCS_Files = GoogleCloudStorageListOperator(
                task_id='GCS_Files',
                bucket=cf.storage.import_bucket_name,
                prefix='20190701/',
                delimiter='.csv',
                dag=dag
            )

for idx, elem in enumerate(["{{ task_instance.xcom_pull(task_ids='GCS_Files') }}"]):
    storage_to_bigquery = GoogleCloudStorageToBigQueryOperator(
            task_id='storage_to_bigquery',
            bucket=cf.storage.import_bucket_name,
            create_disposition='CREATE_IF_NEEDED',
            autodetect=True,
            destination_project_dataset_table=f"{cf.project}.{cf.bigquery.core_dataset_name}.{idx}",
            skip_leading_rows=1,
            source_format='CSV', 
            source_objects=[f'{elem}'],
            write_disposition='WRITE_TRUNCATE',
            dag=dag
            )

    storage_to_bigquery.set_upstream(GCS_Files)

Однако в списке не удается выполнить итерацию по одному, выдавая указанную ниже ошибку.

googleapiclient.errors.HttpError: <HttpError 400 when requesting https://bigquery.googleapis.com/bigquery/v2/projects/my-project/jobs?alt=json returned "Source URI must not contain the ',' character: gs://mybucket/['20190701/file0.csv', '20190701/file1.csv', '20190701/file2.csv']">

Есть предложения?заранее спасибо.

1 Ответ

1 голос
/ 04 июля 2019

Вы не можете вызывать макрос из любого места в вашем коде. Это выглядит как строка в вашем коде: "{{task_instance.xcom_pull (task_ids = 'GCS_Files')}}" И позже вычисляется Jinja2 при передаче в оператор gcp, потому что вы используете шаблонное поле: https://github.com/apache/airflow/blob/21a7e7ec67ac7a391d837aa7c13c0825683f1349/airflow/contrib/operators/gcs_to_bq.py#L140

Чтобы иметь возможность вызывать task_instance.xcom_pull, вам необходим контекст, который может существовать только в прогоне DAG. Когда Airflow лениво оценивает DAG, это недоступно.

В вашем случае лучше всего использовать SubDAG, чтобы зациклить вашего оператора, используя ваш макрос для генерации списка файлов для зацикливания: https://airflow.apache.org/concepts.html#subdags

...