Воздушный поток - попытка зациклить оператора.Выполнение не ожидает фактической операции для завершения

Question

В Airflow - я пытаюсь зациклить оператора.(BigQueryOperator).Группа обеспечения доступности баз данных завершается еще до завершения запроса.

По сути, моя группа обеспечения доступности баз данных:

1. Считывает набор запросов вставки один за другим.
2. запускает каждый запрос с помощью BigQueryOperator.

Когда я пытаюсь написать 2 записи (с 2 операторами вставки) - после задания я вижу только 1 запись.

dag
bteqQueries = ReadFile() --Read GCP bucket file and get the list of SQL queries (as text) separated by new line

for currQuery in bteqQueries.split('\n'):
    #logging.info("currQuery : {}".format(currQuery))
     parameter = {
    'cur_query': currQuery
}
    logging.info("START $$ : {}".format(parameter.get('cur_query')))
    gcs2BQ = BigQueryOperator(
    task_id='gcs2bq_insert',
    bql=parameter.get('cur_query'),
    write_disposition="WRITE_APPEND",
    bigquery_conn_id='bigquery_default',
    use_legacy_sql='False',
    dag=dag,
    task_concurrency=1)
    logging.info("END $$ : {}".format(parameter.get('cur_query')))


gcs2BQ

Ожидается выполнение всех запросов во входном файле (в сегменте GCS).У меня было несколько запросов на вставку и я ожидаю 2 записи в итоговой таблице больших запросов.Но я вижу только 1 запись.

******** Ниже находится журнал ******

 2018-12-19 03:57:16,194] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,190] {gcs2BQ_bteq.py:59} INFO - START $$ : insert into `gproject.bucket.employee_test_stg.employee_test_stg` (emp_id,emp_name,edh_end_dttm) values (2,"srikanth","2099-01-01") ; 
[2018-12-19 03:57:16,205] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,201] {models.py:2190} WARNING - schedule_interval is used for <Task(BigQueryOperator): gcs2bq_insert>, though it has been deprecated as a task parameter, you need to specify it as a DAG parameter instead
[2018-12-19 03:57:16,210] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,209] {gcs2BQ_bteq.py:68} INFO - END $$ : insert into `project.bucket.employee_test_stgemployee_test_stg` (emp_id,emp_name,edh_end_dttm) values (2,"srikanth","2099-01-01") ; 
[2018-12-19 03:57:16,213] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,213] {gcs2BQ_bteq.py:59} INFO - START $$ : insert into `project.bucket.employee_test_stg` (emp_id,emp_name,edh_end_dttm) values (3,"srikanth","2099-01-01") ;
[2018-12-19 03:57:16,223] {base_task_runner.py:98} INFO - Subtask: 
[2018-12-19 03:57:16,218] {models.py:2190} WARNING - schedule_interval is used for <Task(BigQueryOperator): gcs2bq_insert>, though it has been deprecated as a task parameter, you need to specify it as a DAG parameter instead
[2018-12-19 03:57:16,230] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,230] {gcs2BQ_bteq.py:68} INFO - END $$ : insert into `dataset1.adp_etl_stg.employee_test_stg` (emp_id,emp_name,edh_end_dttm) values (3,"srikanth","2099-01-01") ;
[2018-12-19 03:57:16,658] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,655] {bigquery_operator.py:90} INFO - Executing: insert into `dataset1.adp_etl_stg.employee_test_stg` (emp_id,emp_name,edh_end_dttm) values (2,"srikanth","2099-01-01") ; 
[2018-12-19 03:57:16,703] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,702] {gcp_api_base_hook.py:74} INFO - Getting connection using `gcloud auth` user, since no key file is defined for hook.
[2018-12-19 03:57:16,848] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,847] {discovery.py:267} INFO - URL being requested: GET https://www.googleapis.com/discovery/v1/apis/bigquery/v2/rest
[2018-12-19 03:57:16,849] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:16,849] {client.py:595} INFO - Attempting refresh to obtain initial access_token
[2018-12-19 03:57:17,012] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:17,011] {discovery.py:852} INFO - URL being requested: POST https://www.googleapis.com/bigquery/v2/projects/gcp-***Project***/jobs?alt=json
[2018-12-19 03:57:17,214] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:17,214] {discovery.py:852} INFO - URL being requested: GET https://www.googleapis.com/bigquery/v2/projects/gcp-***Project***/jobs/job_jqrRn4lK8IHqTArYAVj6cXRfLgDd?alt=json
[2018-12-19 03:57:17,304] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:17,303] {bigquery_hook.py:856} INFO - Waiting for job to complete : gcp-***Project***, job_jqrRn4lK8IHqTArYAVj6cXRfLgDd
[2018-12-19 03:57:22,311] {base_task_runner.py:98} INFO - Subtask: [2018-12-19 03:57:22,310] {discovery.py:852} INFO - URL being requested: GET https://www.googleapis.com/bigquery/v2/projects/gcp-***Project***/jobs/job_jqrRn4lK8IHqTArYAVj6cXRfLgDd?alt=json

Answer 1

Попробуйте использовать следующий код:

gcs2BQ = []
for index, currQuery in enumerate(bteqQueries.split('\n')):
    logging.info("currQuery : {}".format(currQuery))
    parameter = {
        'cur_query': currQuery
    }
    logging.info("START $$ : {}".format(parameter.get('cur_query')))
    gcs2BQ.append(BigQueryOperator(
        task_id='gcs2bq_insert_{}'.format(index),
        bql=parameter.get('cur_query'),
        write_disposition="WRITE_APPEND",
        bigquery_conn_id='bigquery_default',
        use_legacy_sql='False',
        dag=dag,
        task_concurrency=1))
    logging.info("END $$ : {}".format(parameter.get('cur_query')))

    if index == 0:
        gcs2BQ[0]
    else:
        gcs2BQ[index - 1] >> gcs2BQ[index]

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