> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Rerun Airflow Dags and tasks
You can set when to run Airflow Dags using a wide variety of [scheduling](/docs/learn/scheduling-in-airflow) options. Some use cases where you might want tasks or Dags to run outside of their regular schedule include:
* You want one or more tasks to automatically run again if they fail.
* You need to manually rerun a failed task for one or multiple Dag runs.
* You want to deploy a Dag with a start date of one year ago and trigger all Dag runs that would have been scheduled in the past year.
* You have a running Dag and realize you need it to process data for two months prior to the Dag's start date.
In this guide, you'll learn how to configure automatic retries, rerun tasks or Dags, trigger historical Dag runs, and review the Airflow concepts of catchup and backfill.
## Assumed knowledge
To get the most out of this guide, you should have an understanding of:
* Dag scheduling. See [Schedule Dags in Airflow](/docs/learn/scheduling-in-airflow)
## Automatically retry tasks
In Airflow, you can configure individual tasks to retry automatically in case of a failure. The default number of times a task will retry before failing permanently can be defined at the Airflow configuration level using the core config `default_task_retries`. You can set this configuration either in `airflow.cfg` or with the environment variable `AIRFLOW__CORE__DEFAULT_TASK_RETRIES`.
You can overwrite the `default_task_retries` of an Airflow environment at the task level by using the `retries` parameter.
The `retry_delay` parameter (default: `timedelta(seconds=300)`) defines the time spent between retries. You can set a maximum value for the retry delay in the core Airflow config, `max_task_retry_delay` (`AIRFLOW__CORE__MAX_TASK_RETRY_DELAY`), which, by default, is set at 24 hours. Or, for individual tasks, you can set the maximum retry delay with the parameter, `max_retry_delay`.
To progressively increase the wait time between retries until `max_retry_delay` is reached, set `retry_exponential_backoff` to `True` for the delay to double with each retry. In Airflow 3.2+ you can set `retry_exponential_backoff` to a float to directly specify the factor by which the retry delay should be multiplied between retries. For example, to multiply the retry delay by 3 between retries, set `retry_exponential_backoff` to `3.0`.
It is common practice to set the number of retries for all tasks in a Dag by using `default_args` and override it for specific tasks as needed. To override specific tasks, provide a different value to the task level `retries` parameter.
The Dag below contains 4 tasks that will always fail. Each of the tasks uses a different retry parameter configuration.
```python expandable wrap theme={null}
from airflow.decorators import dag
from airflow.operators.bash import BashOperator
from pendulum import datetime, duration
@dag(
start_date=datetime(2023, 4, 1),
schedule="@daily",
catchup=False,
default_args={
"retries": 3,
"retry_delay": duration(seconds=2),
"retry_exponential_backoff": True,
"max_retry_delay": duration(hours=2),
},
)
def retry_example():
t1 = BashOperator(task_id="t1", bash_command="echo I get 3 retries! && False")
t2 = BashOperator(
task_id="t2",
bash_command="echo I get 6 retries and never wait long! && False",
retries=6,
max_retry_delay=duration(seconds=10),
)
t3 = BashOperator(
task_id="t3",
bash_command="echo I wait exactly 20 seconds between each of my 4 retries! && False",
retries=4,
retry_delay=duration(seconds=20),
retry_exponential_backoff=False,
)
t4 = BashOperator(
task_id="t4",
bash_command="echo I have to get it right the first time! && False",
retries=0,
)
retry_example()
```
### Retry policies
In Airflow 3.3+, rather than applying a fixed retry count as described previously, you can attach a retry policy that determines whether, when, and how a task is retried based on the type of failure that occurs.
For example, for exceptions that you know may be transient (like a 503 error), you can retry the task multiple times with a backoff. For exceptions that require manual intervention (like an authentication issue), you can fail the task immediately.
You can define your retry logic using `ExceptionRetryPolicy`. Each rule maps an exception type to an action, with an optional custom retry delay and a reason that gets logged. You can define the policy in a Dag file directly, or add it to your include/ folder and import it into multiple Dags. Any task errors that aren't covered by your policy will use the default retry behavior that you've set for your task, Dag, or Airflow environment.
```python expandable wrap theme={null}
from datetime import timedelta
from airflow.sdk import (
ExceptionRetryPolicy,
RetryAction,
RetryRule,
task,
)
SNOWFLAKE_RETRY_POLICY = ExceptionRetryPolicy(
rules=[
RetryRule(
exception="snowflake.connector.errors.OperationalError",
action=RetryAction.RETRY,
retry_delay=timedelta(minutes=2),
reason="Transient connection or warehouse issue",
),
RetryRule(
exception="snowflake.connector.errors.ProgrammingError",
action=RetryAction.FAIL,
reason="SQL or compilation error, query edit needed",
),
],
)
@task(
retries=4,
retry_delay=timedelta(seconds=30),
retry_policy=SNOWFLAKE_RETRY_POLICY,
)
def daily_aggregation():
...
```
In this example, the Dag runs queries against Snowflake. If an operational error is raised, the task retries after a two-minute delay. Since this kind of error is usually caused by something like a network blip or an account-level concurrency limit, waiting and trying again makes sense. But if a programming error is raised, the task fails immediately. Programming errors are usually a problem with the SQL itself, so retrying won't produce a different result.
For more on this feature, including exception matching and composition with existing retry parameters, see the [Airflow docs](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/tasks.html#retry-policies).
### Custom retry policies
You can also implement a custom retry policy by subclassing `RetryPolicy` and implementing its `evaluate()` method. This lets you inspect exception attributes or use context that `ExceptionRetryPolicy` doesn't have access to.
Building on the example in the previous section, some database errors can only be distinguished by an attribute on the exception rather than by the exception class itself. A custom policy can implement retry logic on those attributes:
```python wrap theme={null}
from datetime import timedelta
from airflow.sdk import RetryDecision, RetryPolicy
class SnowflakeErrorCodeRetryPolicy(RetryPolicy):
"""Route Snowflake errors by error code rather than exception class."""
RESOURCE_PRESSURE_CODES = {605, 606} # statement canceled, resource shortage
SCHEMA_ERROR_CODES = {2003, 2043, 3001} # missing object, cannot drop, invalid identifier
def evaluate(self, exception, try_number, max_tries, context=None):
errno = getattr(exception, "errno", None)
if errno in self.RESOURCE_PRESSURE_CODES:
return RetryDecision.retry(retry_delay=timedelta(minutes=5))
if errno in self.SCHEMA_ERROR_CODES:
return RetryDecision.fail(reason=f"Snowflake error {errno}, retries won't fix it")
return RetryDecision.default()
```
Another common example would be customizing retry behavior based on how much time has passed since the Dag started running. If your Dag produces time-sensitive data products, you may want the task to fail sooner so the team is notified, rather than waiting on multiple retries.
```python wrap theme={null}
from datetime import timedelta
from airflow.sdk import RetryDecision, RetryPolicy
from airflow.utils import timezone
class StaleRunRetryPolicy(RetryPolicy):
"""Stop retrying once the run is too stale to be useful."""
STALENESS_THRESHOLD = timedelta(hours=6)
def evaluate(self, exception, try_number, max_tries, context=None):
if context is None:
return RetryDecision.default()
logical_date = context["dag_run"].logical_date
if timezone.utcnow() - logical_date > self.STALENESS_THRESHOLD:
return RetryDecision.fail(
reason="Run is past the staleness threshold, escalating"
)
return RetryDecision.default()
```
## Automatically pause a failing Dag
You can configure Airflow to automatically pause a Dag after a certain number of failed Dag runs, preventing a failing Dag from continuing to run and potentially causing more issues.
To set the maximum number of consecutive failed Dag runs for all your Dags in your Airflow environment, set the `core.max_consecutive_failed_dag_runs_per_dag` config. For example, to automatically pause all your Dags after they had 5 failed Dag runs in a row, set:
```text wrap theme={null}
AIRFLOW__CORE__MAX_CONSECUTIVE_FAILED_DAG_RUNS_PER_DAG=5
```
You can override this setting for a specific Dag by setting the `max_consecutive_failed_dag_runs` parameter in the Dag instantiation. For example, to pause a specific Dag after 3 failed Dag runs in a row, set:
TaskFlow
```python wrap theme={null}
# from airflow.sdk import dag
# from pendulum import datetime
@dag(
start_date=datetime(2024, 4, 1),
schedule="@daily",
max_consecutive_failed_dag_runs=3,
catchup=False,
)
def my_dag():
# Define your tasks here
my_dag()
```
Traditional
```python wrap theme={null}
# from airflow.sdk import DAG
# from pendulum import datetime
with DAG(
dag_id="my_dag",
start_date=datetime(2024, 4, 1),
schedule="@daily",
max_consecutive_failed_dag_runs=3,
catchup=False,
):
# Define your tasks here
```
The `max_consecutive_failed_dag_runs` config and Dag-level parameter is currently experimental and might be subject to breaking changes in future releases.
## Manually rerun tasks or Dags
[Rerunning tasks](https://airflow.apache.org/docs/apache-airflow/stable/dag-run.html#re-run-tasks) or full Dags in Airflow is a common workflow.
To rerun a task in Airflow you clear the task status to update the `max_tries` and current task instance state values in the metastore. After the task reruns, the `max_tries` value updates to `0`, and the current task instance state updates to `None`.
To clear the task status, go to your Dag in the Airflow UI, select the task instance you want to rerun and click the **Clear Task Instance** button.
A popup window appears, giving you the following options to clear and rerun additional task instances related to the selected task:
* Past: Clears any instances of the task in Dag runs with a logical date before the selected task instance.
* Future: Clears any instances of the task in Dag runs with a logical date after the selected task instance.
* Upstream: Clears any tasks in the current Dag run which are upstream from the selected task instance.
* Downstream: Clears any tasks in the current Dag run which are downstream from the selected task instance.
* Only Failed: Clears only failed instances of any task instances selected based on the above options.
The window shows which task instances will be cleared with the current settings. Click **Confirm** and the task(s) will be cleared and rescheduled for another run.
You can also use the [Airflow CLI](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#clear) or [API](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/patch_task_instance) to programmatically clear task instances.
To clear a full Dag run, click the Dag run, and then click **Clear Run** as shown in the following image.
Don't clear or change task statuses directly in the Airflow metastore. This can cause unexpected behavior in Airflow.
### Add notes to cleared tasks and Dags
You can add notes to task instances and Dag runs in the Airflow UI. This feature is useful for tracking manual changes to task instances, such as reruns or task status changes. Astronomer recommends leaving a note on a task or Dag whenever you manually update a task instance through the Airflow UI.
To add a note to a task instance or Dag run:
1. Go to your Dag in the Airflow UI.
2. Select a task instance or Dag run.
3. Click **Add a note**.
4. Write a note and click **Confirm**.
We recommend using this feature for tracking and maintaining visibility of manual changes made to task instances such as rerunning or changing the task status.
## Catchup
You can use the built-in [catchup](https://airflow.apache.org/docs/apache-airflow/stable/dag-run.html#catchup) Dag argument to process data for logical dates between the set `start_date` of a Dag and the current date.
When the catchup parameter for a Dag is set to `True`, at the time the Dag is turned on in Airflow, the scheduler starts a Dag run for every data interval that hasn't been run between the Dag's `start_date` and the current data interval. For example, if your Dag is scheduled to run daily and has a `start_date` of 1/1/2025, and you deploy that Dag and turn it on 2/1/2025, Airflow will schedule and start all of the daily Dag runs for January. Catchup is also triggered when you turn a Dag off for a period and then turn it on again.
Catchup can be controlled by setting the parameter in your Dag's arguments. By default, catchup is set to `False`. This example Dag has catchup turned on:
```python wrap theme={null}
@dag(
dag_id="example_dag",
start_date=datetime(2025, 4, 23),
max_active_runs=1,
schedule="@daily",
default_args={
"retries": 1,
"retry_delay": timedelta(minutes=3),
},
catchup=True
)
```
Catchup is a powerful feature, but it should be used with caution. For example, if you deploy a Dag that runs every 5 minutes with a start date of 1 year ago and set catchup to `True`, Airflow will schedule numerous Dag runs all at once. When using catchup, keep in mind what resources Airflow has available and how many Dag runs you can support at one time. To avoid overloading your scheduler or external systems, you can use the following parameters in conjunction with catchup:
* `max_active_runs`: Set at the Dag level and limits the number of Dag runs that Airflow will execute for that particular Dag at any given time. For example, if you set this value to 3 and the Dag had 15 catchup runs to complete, they would be executed in 5 chunks of 3 runs.
* `depends_on_past`: Set at the task level or as a `default_arg` for all tasks at the Dag level. When set to `True`, the task instance must wait for the same task in the most recent Dag run to be successful. This ensures sequential data loads and allows only one Dag run to be executed at a time in most cases.
* `wait_for_downstream`: Set at the Dag level and similar to a Dag-level implementation of `depends_on_past`. The entire Dag needs to run successfully for the next Dag run to start.
If you want to set catchup to True by default for all Dags in your Airflow environment, for example for migration purposes, you can set the Airflow config `AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT` to `True`.
If you want to deploy your Dag with catchup enabled but there are some tasks you don't want to run during the catchup, you can use the [`LatestOnlyOperator`](https://airflow.apache.org/registry/providers/standard#standard-latest_only-LatestOnlyOperator) in your Dag. This operator only runs during the Dag's most recent scheduled interval. In every other Dag run it is ignored, along with any tasks downstream of it.
## Backfill
[Backfilling](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html#backfill) is the concept of running a Dag for a specified period in the past to re-process historical or missed data. Unlike catchup, which triggers missed Dag runs from the Dag's `start_date` through the current data interval, backfill periods can be specified explicitly and can include periods prior to the Dag's `start_date`.
In Airflow 3, backfills are managed by the scheduler and can be triggered through the UI, API, or CLI. To run a backfill from the UI, click the blue **Trigger** button and select **Backfill**. Choose the date range you want to backfill for, and which runs you want to reprocess. You also have the option to select the number of max active runs for the backfill, whether you want to run backwards or forwards, specify run parameters and select one of three reprocessing behaviors:
* **Missing Runs**: Creates and runs only the Dag runs that don't already exist for the selected period.
* **Missing and Errored Runs**: Creates any missing runs and also re-runs any existing Dag runs that previously failed.
* **All Runs**: Clears and re-runs all existing Dag runs within the date range, in addition to creating any that are missing.
Execute the backfill by clicking **Run Backfill**. The UI tells you how many runs will be triggered and the backfill uses the latest [DAG version](/docs/learn/airflow-dag-versioning) that is available for your Dag.
Once the backfill has started, you can pause or cancel it at any time in the UI. Backfilled DAG runs are indicated in your grid by the u-turn arrow.
To see an example of backfilling using the CLI, see the [Airflow docs](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#backfill). For information on how to backfill using the Airflow REST API see the [Airflow REST API docs](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#tag/Backfill).
When using backfill, make sure to consider your available resources. If your backfill will trigger many Dag runs, and/or you have many other Dags running at the same time, you should set a max active runs that won't overload your scheduler.