Порядок выполнения задач в DAG в Google Cloud Composer влияет на то, выполняется ли задача

Question

Я новичок в Google Cloud Composer и столкнулся с тем, что кажется странной проблемой в созданной мной DAG. У меня есть процесс, который берет файл tar.gz из облачного хранилища, разархивирует его как файл .gz, а затем загружает файл .gz в BigQuery. Вчера я попытался добавить новый шаг процесса, который представляет собой вставку из созданного «осколка» в новую таблицу.

Я не мог заставить это работать, пока я не изменил порядок шагов в моем выполнении DAG. В моей DAG у меня есть шаг, который называется "delete_tar_gz_files_op". Когда это было выполнено до «insert_daily_file_into_nli_table_op», вставка никогда не запускалась (в Composer не было сбоев, просто казалось, что он вообще не запускался). Когда я меняю порядок этих двух шагов без других изменений в коде, вставка работает как ожидалось. Кто-нибудь знает, что может быть причиной этого? Я понятия не имею, почему это произойдет, поскольку эти два шага не связаны вообще. Тот делает запрос вставки из одной большой таблицы запросов в другую. Другой удаляет файл tar.gz, находящийся в облачном хранилище.

Мой порядок выполнения dag, который работает:

initialize >>  FilesToProcess >> download_file >> convert_task >> upload_task >> gcs_to_bq >> archive_files_op >> insert_daily_file_into_nli_table_op >> delete_tar_gz_files_op

Часть используемого кода:

    #The big query operator inserts the files from the .gz file into a table in big query.
    gcs_to_bq = GoogleCloudStorageToBigQueryOperator(
    task_id='load_basket_data_into_big_query'+job_desc,
    bucket="my-processing-bucket",
    bigquery_conn_id='bigquery_default',
    create_disposition='CREATE_IF_NEEDED',
    write_disposition='WRITE_TRUNCATE',
    compression='GZIP',
    source_objects=['gzip/myzip_'+process_date+'.gz'],
    destination_project_dataset_table='project.dataset.basket_'+clean_process_date,
    field_delimiter='|',
    skip_leading_rows=0,
    google_cloud_storage_conn_id="bigquery_default",
    schema_object="schema.json",
    dag=dag
    )

    #The created shard is then inserted into basket_raw_nli.basket_nli.  This is a partitioned table which contains only the NLI subtype
    insert_daily_file_into_nli_table_op = bigquery_operator.BigQueryOperator(
    task_id='insert_daily_file_into_nli_table_op_'+job_desc,
    bql=bqQuery,
    use_legacy_sql=False,
    bigquery_conn_id='bigquery_default',
    write_disposition='WRITE_APPEND',
    allow_large_results=True,
    destination_dataset_table=False,
    dag=dag)

    #The tar file created can now be deleted from the raw folder
    delete_tar_gz_files_op=python_operator.PythonOperator(
    task_id='delete_tar_gz_files_'+job_desc,
    python_callable=delete_tar_gz_files,
    op_args=[file, process_date],
    provide_context=False, 
    dag=dag)

    def delete_tar_gz_files(file, process_date):
        execution_command='gsutil rm ' + source_dir + '/' + file     
        print(execution_command)
        returncode=os.system(execution_command)
        if returncode != 0:
            #logging.error("Halting process...")
            exit(1)

Статус ручного запуска: состояние выполнения