What Does Idempotent Mean in Data Pipelines? Preventing Duplicate ETL Runs with Apache Beam

Idempotent means you can run an operation once or many times and get the same result. In a data pipeline, it means a retry, rerun or backfill leaves the target table exactly as one clean run would.

If a retry just doubled your rows, your pipeline wasn't idempotent when it wrote them. Apache Beam retries work automatically and gives exactly-once results inside the pipeline. But it's still up to you to make what your code writes to a database or file safe.

What does idempotent mean?

Idempotent means that doing the same operation many times has the same effect as doing it once. A pipeline run is idempotent when running it twice for the same input window leaves the target table in exactly the same state as running it once.

HTTP has the clearest everyday example. A method is idempotent when many identical requests have the same intended effect on the server as one request. PUT, DELETE and GET are idempotent. POST is not.

The difference is in what each request sends. A PUT body is the final state you want the resource to be in, so sending it again changes nothing. A POST body is information for the server to process, so sending it again can create a second record. Clients can automatically retry idempotent requests after a network failure, since repeating them is harmless. Data loads can behave like either method:

• An overwrite of the partition for 2026-10-12 works like PUT. The second run replaces the rows with the same rows.
• A plain INSERT of that day's rows works like POST. The second run adds a second copy.

Apache Beam expects the same from your code. Its programming guide recommends that each function can be repeated or retried as often as necessary without causing unintended side effects, because Beam doesn't promise how many times your code runs.

Why do ETL pipelines write duplicate rows?

ETL pipelines write duplicate rows when some work runs more than once and the final write appends instead of replacing. Retries and reruns are normal, so the duplicates come from the write, not from the retry.

• Bundle retries inside the runner. Beam splits a collection into bundles. If one element in a bundle fails, the whole bundle fails and must be retried, often on a different worker. Every element in that bundle runs through your code again, including the ones that already succeeded.
• At-least-once delivery. Dataflow streaming jobs can run in an at-least-once mode that costs less and has lower latency, but allows duplicates.
• Scheduler retries. Airflow retries failed tasks, which is why its guide says tasks should produce the same outcome on every re-run and should use UPSERT instead of INSERT.
• Backfills and manual reruns. Someone fixes a bug and reruns last week. Every day that already loaded gets loaded again.

The runner cleans up after its own retries. When a bundle fails, Beam throws away the bundle's output and recomputes it. But that cleanup only covers data inside the pipeline. If your code already sent a row to a database, called an API, or wrote a file, that write stays where it landed, and the retry can repeat it.

[IMAGE 1: upload gtm-1361-img1.png here]

Non-deterministic code makes this worse. If a task reads "the latest data" or calls now(), a rerun processes different rows than the first run did. If you key each run on a fixed partition, such as Airflow's data_interval_start, every rerun gets the same input.

What does Apache Beam guarantee, and where does it stop?

Beam on Dataflow guarantees exactly-once results inside the pipeline: records are processed at least once, and the results are committed exactly once. Writes to an external store and other side effects are only exactly-once if the sink supports it.

On Dataflow, batch pipelines always use exactly-once processing, and streaming pipelines use it by default. But in both cases a single record can still pass through a transform more than once, or even run on two workers at the same time. A DoFn that calls a remote API can make that call twice for the same record, so making the write safe is up to you.

Use a built-in sink with exactly-once writes

For BigQuery, the write method you pick decides what a retry does:

Here's how the BigQueryIO page lists the exactly-once method:

[IMAGE 2: upload gtm-1361-img2.png here]

WRITE_TRUNCATE replaces the whole table. Overwriting a single date partition happens on the warehouse side, which we'll cover in the next section.

BigQuery doesn't enforce primary key uniqueness, so the CDC option only works when the key is already unique in your data.

Freeze the input before a side effect

Idempotency breaks when a retry sends a different set of records. For example, a windowed sum can fire, then fire again with a late element added. If you write that sum twice by key, the row still ends up stable, but a downstream system that already acted on the first value acted on the wrong number.

Adding a Reshuffle before the side effect fixes this. It helps guarantee that only one version of a DoFn's output can make it past a shuffle boundary. In Java you can also mark the writing DoFn with @RequiresStableInput, so a retry sees the same input as the first attempt.

In a streaming job, you put the Reshuffle right before the step that writes to the outside system:

[IMAGE 3: upload gtm-1361-img3.png here]

Drop repeats with Deduplicate

Beam has a Deduplicate transform in Java and Python, but it is best effort and bounded by time. In Python you pass a processing-time or event-time duration so the state doesn't grow forever. Java defaults to 10 minutes of processing time. Values in different windows never count as duplicates.

So Deduplicate is a streaming tool for messages that get delivered again. It does nothing for a backfill that reruns last Tuesday. For batch reruns, the write has to be idempotent.

Which patterns make a load idempotent?

You can make a load idempotent by overwriting the partition, merging on a key, building deterministic IDs, deduping in a staging table, or checking the data before publishing it. Each one turns an append into something closer to a PUT.

[IMAGE 4: upload gtm-1361-img4.png here]

Partition overwrite is the simplest option for daily batch jobs. A run for 2026-10-12 owns that partition. If you run it again, it replaces the partition with the same rows, and a backfill of last week replaces seven partitions. This only works when the run reads a fixed window. That's why Airflow recommends reading and writing a specific partition and never "the latest available data".

Merge is the better fit when updates arrive for old rows, such as an order whose status changes days later. A partition overwrite rebuilds every day that order touched, while a merge on order_id updates one row. The key has to be unique in each batch you merge, so dedupe the batch first if the source can repeat a row.

Deterministic IDs are what make the other patterns work in Beam. If you hash the business key (order ID plus date), every retry gets the same ID. A Beam step can then group on that ID and keep one row, and the warehouse can merge on it.

The two-step patterns, staging and write-audit-publish, add an extra step but are safer. They make sense when a bad load costs more than a slow one.

Example: running a Beam pipeline twice

If you run the same Beam job twice, you can see the difference in a few lines. The append version doubles its rows on the second run, while the idempotent version writes the same two rows both times.

The demo runs locally on the DirectRunner, so you can run it on your own machine. It uses three orders for 2026-10-12, and the source sends order 2 twice. Install Beam first:

pip install apache-beam

[IMAGE 5: upload gtm-1361-img5.png here]

The pipeline has two paths. run_append appends every row to one file from inside a DoFn, which is exactly the side effect the Beam docs warn about. run_idempotent hashes the order ID and date into a key, keeps one row per key, and writes the date's partition file, replacing it on every run:

import hashlib
import json
import shutil
from pathlib import Path

import apache_beam as beam

DS = "2026-10-12"
OUT = Path(__file__).parent / "out"
ROWS = [
{"order_id": 1, "ds": DS, "amount": 10},
{"order_id": 2, "ds": DS, "amount": 20},
{"order_id": 2, "ds": DS, "amount": 20},
]


def encode_row(row):
return json.dumps(row, sort_keys=True)


class AppendRow(beam.DoFn):
def process(self, row):
with (OUT / "append.jsonl").open("a", encoding="utf-8") as output:
output.write(encode_row(row) + "\n")
return []


def matches_partition(row, ds):
return row["ds"] == ds


def key_row(row):
identity = f"{row['order_id']}|{row['ds']}"
key = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:16]
return key, row


def pick_row(key_and_rows):
_, rows = key_and_rows
return min(rows, key=encode_row)


def run_append():
with beam.Pipeline(runner="DirectRunner") as pipeline:
pipeline | beam.Create(ROWS) | beam.ParDo(AppendRow())


def run_idempotent(ds):
with beam.Pipeline(runner="DirectRunner") as pipeline:
(
pipeline
| beam.Create(ROWS)
| beam.Filter(matches_partition, ds)
| beam.Map(key_row)
| beam.GroupByKey()
| beam.Map(pick_row)
| beam.Map(encode_row)
| beam.io.WriteToText(
str(OUT / f"ds={ds}" / "part"),
shard_name_template="",
file_name_suffix=".jsonl",
)
)


def count_rows(path):
with path.open(encoding="utf-8") as source:
return sum(1 for _ in source)


def main():
shutil.rmtree(OUT, ignore_errors=True)
OUT.mkdir()
for run in range(1, 3):
run_append()
run_idempotent(DS)
append_count = count_rows(OUT / "append.jsonl")
idempotent_count = count_rows(OUT / f"ds={DS}" / "part.jsonl")
print(
f"run {run}: append rows={append_count} idempotent rows={idempotent_count}"
)


if __name__ == "__main__":
main()

Here's the output from running both jobs twice:

run 1: append rows=3 idempotent rows=2
run 2: append rows=6 idempotent rows=2

The append path gets it wrong twice. It keeps the source's duplicate order on the first run, then copies all three rows again on the second. The idempotent path removes the source duplicate using the hashed key and replaces the partition on the rerun, so the file has the same two rows after both runs.

pick_row sorts the rows for a key and takes the first, so the same input always picks the same row. And WriteToText with an empty shard template writes one file and replaces it, which is a partition overwrite at the file level. On BigQuery, the same keyed rows would go to a merge or a partition overwrite instead of a file.

How does Erathos handle reruns?

Erathos, a managed data ingestion platform, handles reruns through the sync type you pick for each table. Its recommended sync type, Partial Overwrite, works like a merge on the primary key, so a rerun of that job updates the existing rows. The other sync types also line up with the patterns from earlier:

The Erathos docs describe the partial types like this:

[IMAGE 6: upload gtm-1361-img6.png here]

The partial types need a primary key and a date or datetime column to use as the cursor. For APIs, Erathos picks the cursor for you, and for databases you set it per job.

Partial Append is the right choice when you want every version of a record, such as an audit trail of status changes. It is the wrong choice when downstream models expect one row per key.

Cursor-based loads have one gap that idempotency doesn't fix. If a row gets deleted in the source, it stays in the warehouse forever, because the cursor only finds new and updated rows. You can use the Enable Full On setting to schedule regular full refreshes that clean up those differences. For PostgreSQL, Erathos also supports change data capture (CDC). CDC reads the write-ahead log, where PostgreSQL records every insert, update and delete, so deletes reach the warehouse too. You can set it up in the PostgreSQL connector.