Handling Retries and Idempotency in ETL Jobs: A Practical Guide
In this article let us look at two things that separate a production ETL job from one that just happens to work on a Tuesday: retries and idempotency. Most of us have written pipelines that run fine until they do not — a network blip, a throttled API, a schema change that slips through. When you are building data pipelines that run in production, these are not edge cases. They are the normal state of things.
We will walk through what each concept means, patterns you can apply, some code examples using Python and SQL, and the practical limitations you will hit once you move past a demo environment.
What Do We Mean by Retries and Idempotency?
Retries are straightforward: when an operation fails for a transient reason, you try it again. The key word here is transient. If the source file does not exist, no amount of retrying will help you. But if the API returned a 503 because the service is momentarily overloaded, waiting a few seconds and trying again is often all you need.
Idempotency is the property that running the same operation multiple times produces the same result as running it once. If your pipeline reprocesses the same input data three times because of a retry loop, the output should be identical to what you would get from a single clean run.
These two things go hand in hand. Retries without idempotency mean duplicate rows, inflated aggregates, and angry stakeholders. Idempotency without retries means a single transient failure still takes your pipeline down.
A Simple Retry Pattern
Let us start with a basic retry wrapper in Python. This is the kind of thing you can drop into any script without bringing in a heavyweight framework.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import time
import random
import logging
def retry_with_backoff(func, max_attempts=3, base_delay=2):
"""Call func with exponential backoff on failure."""
logger = logging.getLogger(__name__)
for attempt in range(1, max_attempts + 1):
try:
return func()
except Exception as e:
if attempt == max_attempts:
logger.error(f"All {max_attempts} attempts failed: {e}")
raise
delay = base_delay ** attempt + random.uniform(0, 1)
logger.warning(
f"Attempt {attempt} failed ({e}). "
f"Retrying in {delay:.1f}s..."
)
time.sleep(delay)
The jitter — that random.uniform(0, 1) — is important. Without it, multiple workers hitting the same failing API at the same time will keep piling on at exactly the same intervals. With jitter, the load spreads out and the service has a better chance of recovering.
Here is how you would use it around a flaky API call:
1
2
3
4
5
6
7
8
9
10
def fetch_page(page_number):
response = requests.get(
f"https://api.example.com/data?page={page_number}",
timeout=30
)
response.raise_for_status()
return response.json()
# With retries
data = retry_with_backoff(lambda: fetch_page(5), max_attempts=4)
What Should You Actually Retry?
Not every failure is worth retrying. A quick comparison:
| Scenario | Retry? | Why |
|---|---|---|
| 429 Too Many Requests | Yes | The server is telling you to slow down. Back off and try. |
| 500/502/503 | Yes | Transient server-side issue. |
| Network timeout | Yes | Could be a temporary blip. |
| 400 Bad Request | No | Your request is malformed. Retrying will not fix it. |
| 401/403 Unauthorized | No | You have an auth problem, not a transient one. |
| 404 Not Found | Depends | If you are polling for a resource that may not exist yet, retry. If you are fetching something that should definitely be there, fail fast. |
| Schema mismatch | No | A retry will not make the columns line up. |
In practice, wrap your retry logic around the exceptions that indicate transient failures, not a blanket except Exception. Catching everything and retrying is how you turn a 5-second failure into a 5-minute outage while your pipeline keeps hammering a dead endpoint.
Making Your ETL Idempotent
Retries solve the transient-failure piece. Idempotency makes sure that when retries happen — or when an orchestrator re-runs a task from scratch — you do not end up with garbage data.
Pattern 1: Overwrite the Partition
If your ETL writes to a partitioned table, the simplest approach is to overwrite the target partition before writing. In BigQuery:
1
2
3
4
5
6
DELETE FROM `project.dataset.sales_fact`
WHERE partition_date = '2025-11-03';
INSERT INTO `project.dataset.sales_fact`
SELECT * FROM staging_sales
WHERE partition_date = '2025-11-03';
In Spark:
1
2
3
4
df.write \
.mode("overwrite") \
.partitionBy("partition_date") \
.parquet("gs://bucket/sales_fact/")
Only the matching partition gets replaced. If your job runs once, it works. If it runs three times, the partition still contains exactly one copy of the data.
Pattern 2: Upsert / Merge
When you cannot just blow away a whole partition — maybe you are processing CDC (change data capture) or incremental updates — you need a merge.
1
2
3
4
5
6
7
8
MERGE INTO target t
USING source s
ON t.id = s.id AND t.partition_date = '2025-11-03'
WHEN MATCHED THEN
UPDATE SET t.amount = s.amount, t.updated_at = s.updated_at
WHEN NOT MATCHED THEN
INSERT (id, amount, partition_date, updated_at)
VALUES (s.id, s.amount, s.partition_date, s.updated_at);
Because the merge is keyed on the business ID, running it twice gives you the same result. The matched rows get updated to the same values, the unmatched rows get inserted only once.
Pattern 3: Idempotency Keys
When calling external APIs or writing to a system that does not natively support idempotency, generate a deterministic key and check before writing.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import hashlib
import json
def idempotency_key(record):
"""Generate a stable key from the record contents."""
content = json.dumps(record, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
def write_if_not_exists(cursor, key, record):
cursor.execute(
"SELECT 1 FROM processed_keys WHERE idempotency_key = %s",
(key,)
)
if cursor.fetchone():
return # Already processed, skip
cursor.execute(
"INSERT INTO target_table (col1, col2) VALUES (%s, %s)",
(record["col1"], record["col2"])
)
cursor.execute(
"INSERT INTO processed_keys (idempotency_key) VALUES (%s)",
(key,)
)
This pattern adds an extra write per record, so it is not free. But it is the most reliable option when the destination has no native upsert or partition-overwrite semantics.
Where This Gets Tricky in Production
1. Side Effects Outside the Database
If your pipeline sends emails, publishes to a message queue, or calls a webhook, retries can be dangerous. A retried task that already sent a notification will send it again. The fix is to separate the data processing from the side effects. Write the data first — idempotently — and only fire side effects after the write is confirmed. Use an outbox pattern or a separate process that reads from your idempotent output.
2. Exactly-Once Semantics Are Harder Than They Sound
Many systems claim exactly-once delivery (Kafka transactions, GCP Dataflow), but they come with performance trade-offs and strict constraints on what your pipeline can do. In practice, most teams settle for at-least-once delivery plus idempotent writes. It is simpler, faster, and easier to reason about.
3. Stateful Retries Across Orchestrator Boundaries
If you use Airflow, Dagster, or Step Functions, the orchestrator has its own retry settings. Your task code also has retries. If both layers retry independently, you get a multiplier effect — 3 orchestrator retries times 3 code-level retries equals 9 attempts per record. Decide which layer owns retry logic and configure the other to fail fast.
4. Deduplication Windows
The processed_keys table from Pattern 3 grows forever. You need a retention policy. In most cases, keeping keys for 7 to 30 days is enough — if the same data has not reappeared in a month, it is probably not a retry anymore, it is a re-ingest. Truncate old keys on a schedule.
5. Testing Retry and Idempotency Logic
It is surprisingly hard to test this stuff. You cannot just run the pipeline end to end and call it done — the happy path works on day one. You need to inject failures deliberately:
- Simulate a network timeout halfway through a batch.
- Kill the process mid-write and restart.
- Run the same input file twice and check for duplicate rows.
If your pipeline passes these three tests, you can sleep better.
Bringing It Together
Here is a sketch of what a production ETL step looks like when you build retries and idempotency in from the start:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def load_partition(date_str):
"""Load one day of data into the fact table, idempotently."""
logger.info(f"Loading partition {date_str}")
# 1. Extract with retries
raw = retry_with_backoff(
lambda: extract_from_source(date_str),
max_attempts=4
)
# 2. Transform (no side effects, safe to retry)
transformed = transform(raw)
# 3. Load — overwrite partition so it's idempotent
retry_with_backoff(
lambda: load_to_bigquery(transformed, date_str, mode="overwrite"),
max_attempts=3
)
logger.info(f"Partition {date_str} loaded successfully")
Nothing fancy. But it handles transient failures without creating duplicates, and that covers the majority of real-world ETL reliability problems.
Wrapping Up
Retries and idempotency are not optional when your pipeline runs on a schedule and someone else is waking up at 3 AM if it fails. The patterns here are simple enough to add to any existing job — pick the idempotency approach that fits your destination (partition overwrite is the easiest if you can use it), wrap your external calls in a retry with backoff and jitter, and separate side effects from data writes. Getting these three things right will catch more production issues than any amount of clever architecture.