Data Pipeline Idempotency | The Garnet Wiki

Data pipeline idempotency is the property of a pipeline operation such that re-executing it after a failure, a manual retry, or an automated recovery cycle produces the same final state as a single successful execution, without accumulating duplicate records, corrupting state, or violating downstream expectations. It matters whenever a pipeline must recover from crashes, handle partial failures, support backfill jobs, or reconcile state after a scheduler restart. Without idempotency, a single retry can double‑count records, break downstream aggregates, or leave the system in a state that requires manual repair.

1. Idempotent writes to a relational store

A common pattern for making a write idempotent is to use a conflict‑resolution clause that checks a business‑unique key before inserting. If the key already exists, the operation either does nothing or applies the same logical update, ensuring that repeated executions leave the row in the same state.

In a psql session, the statement takes the form:

INSERT INTO page_views (event_id, ts, user_id, page)
VALUES ('evt-8739', NOW(), 42, '/home')
ON CONFLICT (event_id) DO NOTHING;

Equivalently, to update the same columns should the row already exist:

INSERT INTO page_views (event_id, ts, user_id, page)
VALUES ('evt-8739', NOW(), 42, '/home')
ON CONFLICT (event_id) DO UPDATE
  SET ts = EXCLUDED.ts,
      user_id = EXCLUDED.user_id,
      page = EXCLUDED.page;

The ON CONFLICT (event_id) clause requires that event_id be the primary key or have a unique index. Without that constraint, the database never encounters a conflict and the INSERT succeeds on every run, silently appending duplicate rows. A frequent pitfall is declaring a surrogate serial primary key while treating another column as the logical unique identifier; the database enforces the surrogate, not the business key, and replays corrupt the dataset.

To execute the statement from a shell script:

psql -h localhost -U dw_user -d warehouse -c \
"INSERT INTO page_views (event_id, ts, user_id, page)
VALUES ('evt-8739', NOW(), 42, '/home')
ON CONFLICT (event_id) DO NOTHING;"

2. Retrying API‑backed steps with idempotency keys

When a pipeline step calls an external service—such as an ingestion endpoint, a payment processor, or a notification service—a retry without safeguards can re‑execute side effects. Many REST APIs accept an Idempotency-Key header; the server stores the key alongside the first response and returns the cached result for any subsequent request that carries the same key, without re‑executing the underlying operation.

Using curl, a pipeline might send:

curl -s -X POST https://ingest.service/api/events \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: run-$(date +%s)-$(uuidgen)" \
  -d @event.json

The Idempotency-Key value must be stable for the duration of a logical execution. If the key is regenerated on every retry (e.g., uuidgen invoked inside the script each time the orchestrator re‑runs the step), the server sees a new key and processes the request again, defeating the purpose. The key should be derived from a run identifier that the orchestration system already tracks—such as an Airflow run_id, a Prefect flow_run_id, or a database sequence number—and persisted to a metadata table or environment variable before the HTTP call is made. Some teams embed the key as a UUID generated once at pipeline start and passed as an argument to each subsequent task step.

If the downstream service does not natively support an idempotency key, the pipeline can implement client‑side dedup: hash the request payload, store the hash in a temporary table or cache, and skip the HTTP call if the hash already exists. The fenced code block below illustrates a shell‑based check using psql and curl:

HASH=$(sha256sum event.json | awk '{print $1}')
EXISTS=$(psql -t -A -c "SELECT 1 FROM pipeline_run_keys WHERE run_key = '$HASH';")
if [ -z "$EXISTS" ]; then
  curl -s -X POST https://ingest.service/api/events \
    -H "Content-Type: application/json" \
    -H "Idempotency-Key: $HASH" \
    -d @event.json
  psql -c "INSERT INTO pipeline_run_keys (run_key) VALUES ('$HASH');"
else
  echo "Skipping duplicate request, key already processed."
fi

3. Orchestration‑level retries and task design

Most workflow engines provide a retries configuration that re‑invokes a task’s command after a failure or a timeout. If the task’s body is itself idempotent (as ensured by the patterns above), the orchestrator’s retry is safe. If the task is not idempotent, each retry compounds the damage—writes multiply, APIs charge twice, pipelines diverge.

A practical habit is to pass a stable --run-id argument to every task command, which downstream steps and idempotency keys can reference. For example, a Bash‑based ETL step might be invoked as:

my-etl-process --run-id "${AIRFLOW_RUN_ID:-$(date +%Y%m%d-%H%M%S)}" \
  --step process_events

Inside the process, the script exports RUN_ID and uses it to construct the idempotency key for any external API call, and prepends it to COPY or INSERT statements so the database can match rows across retries. Without an explicit run identifier, each retry may inhabit a different context, making it nearly impossible to guarantee exactly‑once behavior.

4. Stream replay and aggregate safety

Pipelines that consume from a log (Kafka, Kinesis, Pulsar, etc.) and write aggregated results must ensure that a replay of a consumed partition does not double‑count into downstream materialized views or summary tables. The safest pattern is to carry the source event’s unique identifier into the aggregation key and apply the upsert logic from section 1 whenever a new event is processed.

If the pipeline materializes a daily total per user, the SQL upsert might look like:

INSERT INTO daily_totals (dt, user_id, total_events, total_bytes)
VALUES (CURRENT_DATE, 1024, 1, 2048)
ON CONFLICT (dt, user_id) DO UPDATE
  SET total_events = daily_totals.total_events + EXCLUDED.total_events,
      total_bytes = daily_totals.total_bytes + EXCLUDED.total_bytes;

Because the conflict target (dt, user_id) is unique, re‑processing the same event adds its contributions once and no more. If the pipeline simply INSERTs without conflict resolution, every replay of the same event increments the counters, and the aggregate drifts away from the true value. The same principle applies whether the aggregation is performed in SQL, a stream processing framework, or a batch job that writes Parquet files to a data lake.

Sharp edges

Missing unique constraints. A pipeline may use ON CONFLICT DO UPDATE confidently, yet the target table lacks a unique index on the conflict column. PostgreSQL will raise an error at statement time, but many other engines silently fall back to a plain insert, producing duplicates that are difficult to detect without a separate dedup pass.
Timestamp‑based keys. Some pipelines generate idempotency keys from a timestamp (e.g., run-$(date +%s)). If the scheduler retries after the second boundary changes, the key differs, and the server processes the request again. The key must be tied to a run‑level identifier, not a wall‑clock value.
Retry‑generation of keys. As noted, invoking a random UUID generator inside the task body on each retry creates a new key each time. The idempotency contract is only as strong as the key’s stability.
At‑least‑once vs exactly‑once. Orchestrators typically guarantee at‑least‑once delivery: a task may execute zero, one, or many times. Idempotency is the application‑level guarantee that any number of executions yields the same result. Believing the scheduler provides exactly‑once semantics often leads to designs that omit idempotency logic, which breaks as soon as a network partition or pre‑empt‑node kill triggers a retry.
External state without coordination. A task may write to a relational store using an upsert (safe) and simultaneously call an external webhook that is not idempotent (unsafe). The combined effect is non‑deterministic: the database state is correct, but the external system has processed the event twice. Designers must either make the webhook idempotent (by accepting and ignoring duplicate keys) or sequence the operations so that the database write is the single source of truth, and the webhook is triggered only after a successful commit.

What you learn after it breaks

In a production incident several years ago, an ETL job that had just been fail‑over‑recovered wrote a COPY command to a cloud data lake. The scheduler re‑ran the job after a transient outage, and because the COPY target lacked a partition‑level unique identifier, the lake accumulated triplicate partitions of the same day’s events. Downstream dashboards showed event counts that were three times the expected volume, and the data quality team spent two days reconcilating the partitions by writing an ad‑hoc deduplication job that hashed every record against the source’s primary key before overwriting the lake location. Since that event, every pipeline step in the organization is annotated with an idempotency contract in its metadata: either it uses an upsert with a business‑unique key, it publishes an idempotency key to downstream consumers, or it is marked as “not retryable” in the orchestrator with a clear manual‑recovery path. The contract is checked as part of the CI pipeline for data‑engineering changes, and the absence of a contract causes the merge to be blocked until the gap is filled.

This page was rewritten on 16 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.