An Airflow task is idempotent when rerunning it with the same inputs has no additional effect on the overall result. This matters during backfills, manual reruns, and retries: a task may have written data before failing, then execute again. The rerun should not duplicate data or leave other unwanted side effects. section 3.7.2 section 12.2.1
Overwrite one day instead of appending it
Suppose a task fetches one day of events. Writing that day's JSON to a date-specific file and overwriting the file on rerun preserves the result, assuming the API returns the same events. Appending the events to one shared JSON file instead duplicates that day on every rerun. In the Airflow UI, Clear Task provides a useful test: it reruns the task with the same schedule interval values rather than creating a new manual run. section 3.7.2
Listing 3.18 Existing implementation for fetching events
fetch_events = BashOperator(
task_id="fetch_events",
bash_command=(
"mkdir -p /data/events && "
"curl -o /data/events/{{ logical_date | ds }}.json "
"'http://events-api:8081/events/range?"
"start_date={{data_interval_start | ds}}&" #1
"end_date={{data_interval_end | ds}}'" #1
),
)
The fetch_events task uses the logical date in its output filename and the data interval to select events. Inspect how a rerun targets the same file. section 3.7.2
Choose a repeatable write strategy for each destination:
- For files or time-partitioned datasets, overwrite the destination or the affected partition; alternatively, check for an existing result before writing. section 3.7.2 section 13.6
- For database rows, consider an upsert. If inserting a whole batch with pandas, identify the batch by its Airflow execution date, then delete its previous rows and insert its replacement within one transaction. section 3.7.2 section 13.6
- Review effects beyond the main output, such as notifications or state left by a container. An overwritten dataset alone does not make every side effect repeatable. section 12.2.1 section 8.3.3
Listing 13.17 Writing a pandas DataFrame to a SQL database in one operation
--CREATE TABLE citi_bike_rides(
-- tripduration INTEGER,
-- starttime TIMESTAMP,
-- start_location_id INTEGER,
-- stoptime TIMESTAMP,
-- end_location_id INTEGER,
-- airflow_execution_date TIMESTAMP
--);
df = pd.read_csv(... citi bike data ...)
df["airflow_execution_date"] = pd.Timestamp( #1
context["data_interval_start"].timestamp(), #1
unit='s', #1
) #1
engine = sqlalchemy.create_engine(
BaseHook.get_connection(self._postgres_conn_id).get_uri()
)
with engine.begin() as conn: #2
conn.execute(
"DELETE FROM citi_bike_rides"
f"WHERE airflow_execution_date=
➥'{context['data_interval_start']}';"
) #3
df.to_sql("citi_bike_rides", con=conn, index=False,
➥ if_exists="append")
The database example adds an execution-date column and uses a transaction for batch replacement. The transaction also protects against leaving a partly replaced batch after failure. section 13.6
Idempotency concerns the result of repeated execution; atomicity concerns whether a series of operations completes entirely or not at all. Both help make Airflow tasks safe to rerun and backfill. If a task's inputs or code change, assess the new result separately rather than assuming a rerun will reproduce the old one. section 3.8 section 12.2.1
