как установить два DAG в потоке воздуха с помощью ExternalTaskSensor?

Question

У меня есть два DAG:

DAG_CPS

dag = DAG(
  'DAG_CPS',
  default_args=default_args,
  dagrun_timeout=timedelta(hours=2),
  schedule_interval=None,
  max_active_runs=1
) 
tmp1_cap_pes_sap = PostgresOperatorWithTemplatedParams(
task_id='tmp1_cap_pes_sap',
sql='./SQL/A2050.sql',
postgres_conn_id='xxxx',
dag=dag) 
...

DAG_SAS

dag = DAG(
'DAG_SAS',
default_args=default_args,
dagrun_timeout=timedelta(hours=2),
schedule_interval=None,
max_active_runs=1
)

wait_for_DAG_CPS = ExternalTaskSensor(
task_id='wait_for_DAG_CPS',
external_dag_id='DAG_CPS',
external_task_id='tmp1_cap_pes_sap',
execution_delta=None,
execution_date_fn=None,
dag=dag)

Я запускаю обе группы доступности базы данных вручную из Интернета, задача tmp1_cap_pes_sap заканчивается нормально

Attribute       Value
dag_id          DAG_CPS
duration        None
end_date        2018-08-24 11:04:28.177221
execution_date  2018-08-24 11:04:18.113031

но в DAG_SAS я получаю следующий журнал, и он никогда не запускается

[2018-08-24 11:03:55,592] {base_task_runner.py:98} INFO - Subtask: [2018-08-24 11:03:55,592] {sensors.py:243} INFO - Poking for DAG_CPS.tmp1_cap_pes_sap on 2018-08-24T11:03:50.518595 ... 
[2018-08-24 11:04:55,642] {base_task_runner.py:98} INFO - Subtask: [2018-08-24 11:04:55,641] {sensors.py:243} INFO - Poking for DAG_CPS.tmp1_cap_pes_sap on 2018-08-24T11:03:50.518595 ... 
[2018-08-24 11:05:55,718] {base_task_runner.py:98} INFO - Subtask: [2018-08-24 11:05:55,717] {sensors.py:243} INFO - Poking for DAG_CPS.tmp1_cap_pes_sap on 2018-08-24T11:03:50.518595 ... 
[2018-08-24 11:06:55,799] {base_task_runner.py:98} INFO - Subtask: [2018-08-24 11:06:55,797] {sensors.py:243} INFO - Poking for DAG_CPS.tmp1_cap_pes_sap on 2018-08-24T11:03:50.518595 ... 
[2018-08-24 11:07:55,853] {base_task_runner.py:98} INFO - Subtask: [2018-08-24 11:07:55,853] {sensors.py:243} INFO - Poking for DAG_CPS.tmp1_cap_pes_sap on 2018-08-24T11:03:50.518595 ... 

что не так в моем коде?

решаемые

спасибо @Alessandro Cosentino за помощь. Это код после исправления, в принципе он никогда не будет работать, если я запустлю DAG вручную

DAG_CPS

default_args = {
'depends_on_past': False,
'start_date': airflow.utils.dates.days_ago(2),
'retries': 2,
'retry_delay': timedelta(minutes=1)
}

dag = DAG(
'DAG_CPS',
default_args=default_args,
dagrun_timeout=timedelta(minutes=5),
schedule_interval='*/10 * * * *',
max_active_runs=1
)

DAG_SAS

dag = DAG(
'DAG_SAS',
default_args=default_args,
dagrun_timeout=timedelta(minutes=5),
schedule_interval='*/10 * * * *',
max_active_runs=1
)

Answer 1

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

Попробуйте запустить ихвместо этого по тому же графику и посмотрите, работает ли он.

Я предполагаю, что это проблема, из-за существования аргументов execution_delta и execution_date_fn, которые фактически синхронизируют два DAG.См. документы о поведении этих двух аргументов.