Comparing Current and Previous DAG Run States in Apache Airflow: Achieving Conditional DAG Execution

Welcome to the discussion! It appears you’re exploring methods to manage DAG runs in Apache Airflow, specifically aiming to skip subsequent runs based on the status of previous ones. This is a common challenge when orchestrating complex workflows where overlapping or redundant executions need to be avoided. Let’s delve into how you can achieve conditional DAG execution by comparing the current run with the previous one, and explore the limitations and solutions within Airflow.

Understanding DAG Run Behavior and Limitations

You’ve correctly identified catchup and max_active_runs as relevant Airflow settings. Let’s clarify how these parameters, along with wait_for_downstream, influence DAG scheduling and execution, particularly when you want to prevent DAG runs under certain conditions related to previous runs.

• catchup: When catchup is set to False for a DAG, it prevents Airflow from creating backfilled DAG runs for past missed schedules. However, it primarily focuses on preventing past runs, not necessarily skipping future runs based on the current or previous run’s state in real-time. In conjunction with max_active_runs, it does play a role in limiting concurrent executions.

• max_active_runs: Setting max_active_runs to 1 ensures that only one instance of a DAG can be actively running at any given time. While this prevents overlapping runs in the traditional sense, it doesn’t inherently skip the next scheduled run if the current one is still running or has recently failed. If a DAG run is still in progress when the next schedule interval arrives, Airflow, with catchup=False and max_active_runs=1, will indeed skip the runs that were missed during the execution of the active run. However, this is more of a consequence of preventing concurrency than a direct mechanism for conditional skipping based on state.

• wait_for_downstream: This parameter, when set to True at the DAG level or task level, makes downstream tasks in the current DAG run wait for the upstream tasks of the previous DAG run to succeed. It’s designed to manage dependencies across DAG runs, ensuring data consistency or order of operations between different executions. Crucially, wait_for_downstream doesn’t prevent the current DAG run from being created. It only affects the execution of downstream tasks within the current run, based on the previous run’s upstream task status. This isn’t the solution for skipping the entire next DAG run.

Addressing Scenarios for Skipping the Next DAG Run

The core requirement is to skip the next scheduled DAG run under these conditions:

1. The current DAG run is still running when the next schedule time arrives.
2. The current DAG run failed after the next schedule time.
3. The current DAG run failed before the next schedule time.

As noted earlier, scenarios 1 and 2 are partially addressed by setting catchup=False and max_active_runs=1. Missed runs during an active execution window are skipped. However, scenario 3, where a previous run fails before the next scheduled time, presents a different challenge. Airflow’s default scheduler will still create a new DAG run in this case.

Custom Solutions: Sensors for State Comparison

To achieve true conditional skipping based on the previous DAG run’s state, especially for scenario 3, you’ll likely need to go beyond the standard configurations. The most robust approach involves using a custom sensor.

Custom Sensor Approach

A custom sensor allows you to define specific logic to check the state of the most recent DAG run before allowing a new run to proceed. This sensor can be placed at the beginning of your DAG. Here’s the general idea:

1. Query Airflow Metadata Database: Within the sensor, you would query the Airflow metadata database (e.g., using Airflow’s ORM) to retrieve information about the most recent DAG run for the current DAG.
2. State Evaluation: The sensor would evaluate the state of this previous DAG run. You can check for states like running, failed, or success.
3. Conditional Logic: Based on the state, the sensor determines whether to proceed (i.e., allow the current DAG run to continue) or to effectively “skip” the run (by continuously deferring or failing the sensor task, thus preventing downstream tasks from executing).

Challenges with Custom Sensors:

• Complexity: Implementing a custom sensor requires Python coding and familiarity with Airflow’s internal APIs and database schema.
• Database Interaction: Directly querying the metadata database can introduce dependencies and might be less performant than using Airflow’s intended abstractions.
• Maintenance: Custom code requires ongoing maintenance and testing, especially when Airflow versions are upgraded.

Leveraging ExternalTaskSensor

Another, potentially simpler, approach is to utilize the built-in ExternalTaskSensor. While primarily designed to sense the state of tasks in other DAGs, it can be adapted to check the state of tasks in previous runs of the same DAG.

How ExternalTaskSensor Can Help:

You can configure ExternalTaskSensor to look at a specific task in the previous DAG run using the execution_delta parameter. execution_delta allows you to specify a time difference to target a DAG run relative to the current run’s execution date.

Example using ExternalTaskSensor:

from airflow.sensors.external_task_sensor import ExternalTaskSensor
from airflow.utils.state import State
from datetime import timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
}

dag = DAG(
    'conditional_dag_run',
    default_args=default_args,
    schedule_interval=timedelta(minutes=30),
    catchup=True, # Important for execution_delta to work reliably
    max_active_runs=1
)

start_task = DummyOperator(task_id='start', dag=dag)

previous_dagrun_sensor = ExternalTaskSensor(
    task_id='check_previous_dagrun',
    external_dag_id='conditional_dag_run', # Referencing the same DAG
    external_task_id='start', # Sensing a task in the previous run (e.g., the start task)
    allowed_states=[State.SUCCESS], # Define success criteria
    execution_delta=timedelta(minutes=30), # Look back 30 minutes (assuming 30 min schedule)
    dag=dag
)

task_after_check = DummyOperator(task_id='task_after_check', dag=dag)

previous_dagrun_sensor >> task_after_check
start_task >> previous_dagrun_sensor

Explanation:

• catchup=True is crucial here. execution_delta relies on having past DAG runs to reference. With catchup=True, Airflow ensures there’s a “previous” run to check against.
• external_dag_id='conditional_dag_run' and external_task_id='start' : We are telling the sensor to look at the ‘start’ task within the same DAG (conditional_dag_run).
• execution_delta=timedelta(minutes=30): This is set to the schedule interval. The sensor will look for a DAG run that executed approximately 30 minutes before the current run’s scheduled time.
• allowed_states=[State.SUCCESS]: The sensor will only succeed (and allow downstream tasks to proceed) if the ‘start’ task in the previous DAG run was in a ‘success’ state. You can adjust allowed_states and potentially add failed_states for more complex logic.

Limitations of ExternalTaskSensor Approach:

• catchup=True Requirement: Relying on catchup=True might have implications for backfilling behavior if that’s not desired in your overall DAG strategy.
• Sensor Logic Complexity: While simpler than a fully custom sensor, configuring ExternalTaskSensor for this specific use case still requires careful consideration of execution_delta, allowed_states, and task selection.
• Not True “Skipping”: Technically, the DAG run is still created and the sensor task runs. It’s not a complete “skip” at the DAG run level, but it effectively prevents the rest of the DAG from executing if the sensor conditions are not met.

Conclusion and Recommendations

While Airflow doesn’t offer a direct, out-of-the-box “skip next run if previous run is…” feature, you can achieve this conditional execution behavior.

• For scenarios 1 and 2 (preventing overlaps), catchup=False and max_active_runs=1 are effective.
• For scenario 3 and more granular control based on previous run state, a custom sensor provides the most flexibility but demands more development effort.
• ExternalTaskSensor offers a middle ground, allowing you to check previous run states with less custom code, but requires careful configuration and understanding of its behavior, especially the catchup=True dependency when using execution_delta in this way.

Choose the solution that best balances your requirements for control, complexity, and maintainability within your Airflow environment. Remember to thoroughly test any custom sensor or ExternalTaskSensor implementation to ensure it behaves as expected in various DAG run scenarios.