如果下游任务在气流中失败，如何重新运行上游任务（使用Sub Dags）

Question

我有一个气流数据采集器，可以提取数据并执行验证。如果验证失败，则需要重新运行提取。如果验证成功，则继续。

我读过一些人说，潜逃可以解决这个问题，但是我看不到任何例子。我已经尝试过使用sub dag，但是遇到了与尝试在一个DAG中进行操作相同的问题。

如果其中一项失败，我如何让Sub DAG中的所有任务重新运行？

我有以下DAG / sub dag详细信息：

maindag.py

default_args = {
  'owner': 'airflow',
  'depends_on_past': False,
  'start_date': start_date,
  'retries': 3,
  'retry_delay': timedelta(minutes=5),
  'sla': timedelta(hours=sla_hours)
}

main_dag = DAG(
  dag_id,
  default_args=default_args,
  schedule_interval='30 14 * * *',
  max_active_runs=1,
  concurrency=1)

task1 = BashOperator(...)

task2 = SubDagOperator(
  task_id=sub_dag_task_id,
  subdag=sub_dag(dag_id, sub_dag_task_id, start_date, main_dag.schedule_interval),
  dag=main_dag)

task3 = BashOperator(...)

subdag.py

def sub_dag(parent_dag_name, task_id, start_date, schedule_interval):
  dag = DAG(
    '%s.%s' % (parent_dag_name, task_id),
    schedule_interval=schedule_interval,
    start_date=start_date,
    )

  task1 = BashOperator(...)

  task2 = BashOperator(...)

  task3 = BashOperator(...)

  task1 >> task2 >> task3

  return dag

在sub dag中，如果任务3失败，即使任务1成功执行，我也希望它再次运行。为什么这么难做？！

Answer 1

我通过在主dag中创建重试回调方法找到了解决方案：

（原始来源：https : //gist.github.com/nathairtras/6ce0b0294be8c27d672e2ad52e8f2117 ）

from airflow.models import DagBag

def callback_subdag_clear(context):
    """Clears a subdag's tasks on retry."""
    dag_id = "{}.{}".format(
        context['dag'].dag_id,
        context['ti'].task_id
)
    execution_date = context['execution_date']
    sdag = DagBag().get_dag(dag_id)
    sdag.clear(
        start_date=execution_date,
        end_date=execution_date,
        only_failed=False,
        only_running=False,
        confirm_prompt=False,
        include_subdags=False)

然后对于运行subdagoperator的任务，它具有：

on_retry_callback=callback_subdag_clear,

现在，它将清除每个任务的任务实例历史记录，并重新运行子dag中的每个任务，直到主dag中的重试次数为止。