Hashing for idempotency

Every pipeline will run twice on the same data one day. A job times out after writing half its rows and the scheduler retries it. A message queue delivers a message again because the acknowledgement got lost. A colleague re-runs yesterday’s backfill to be safe.

If running twice produces twice the rows, every one of those is a silent bug. The fix has a name: idempotency. An operation is idempotent if doing it twice leaves the same result as doing it once.

You get there by giving every record a stable identity, and the cheapest stable identity is a hash of its content.

Duplicates in the wild

The ridelog dataset has this problem built in. About 5% of app events were ingested twice: same user, same event type, same timestamp, but a new event_id and a later ingested_at. The ingest system minted a fresh id on the retry, so the id cannot tell you they are the same event.

Count them by the fields that describe what happened, not the fields the pipeline added:

SELECT count(*) AS rows,
     count(DISTINCT event_id) AS distinct_ids,
     count(DISTINCT md5(concat_ws('|', user_id, event_type, event_ts))) AS distinct_events
FROM events;

3,146 rows and 3,146 distinct ids, but only 2,976 distinct events. 170 rows are copies. An id generated at ingest time is unique per attempt. What you want is a key that is unique per fact.

What a hash function gives you

A hash function takes any number of bytes and returns a fixed-size number. Two properties matter here:

1. Deterministic. The same input bytes always give the same output, on any machine, in any year.
2. Spread out. Change one byte of the input and the output changes completely, so different records almost never share a hash.

SHA-256 gives 32 bytes, usually written as 64 hex characters. The chance that two different records collide is so small you can ignore it: you would need around 2^128 records before it becomes likely. MD5 is weaker against attackers, but still fine for dedup keys, and many SQL engines have it built in.

The catch is the word bytes. A Python dict is not bytes. You have to turn it into bytes the same way every time.

Canonical bytes first

Two dicts can be equal and still serialise differently:

import hashlib
import json

a = {"user_id": 3, "event_type": "app_open", "event_ts": "2025-03-04 10:51:00"}
b = {"event_ts": "2025-03-04 10:51:00", "event_type": "app_open", "user_id": 3}

def naive(rec):
  return hashlib.sha256(json.dumps(rec).encode()).hexdigest()[:16]

def canonical(rec):
  text = json.dumps(rec, sort_keys=True, separators=(",", ":"))
  return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]

print("equal as dicts:", a == b)
print("naive:     ", naive(a), naive(b))
print("canonical: ", canonical(a), canonical(b))

The naive hash follows insertion order, so the two equal records get different keys. The canonical version fixes three things: keys are sorted, whitespace is fixed, and the text is encoded as UTF-8 on purpose. Now equal records always give equal bytes, and equal bytes always give equal hashes.

Two more rules make this hold up in production:

• Hash only the fields that define the fact. Leave out event_id, ingested_at, file names and load timestamps. They change on retry, and that is the whole problem.
• Normalise values first. "2025-03-04T10:51:00Z" and "2025-03-04 10:51:00" may be the same instant, but they are different bytes. Pick one format before hashing.

An idempotent loader

With a stable key, the loader becomes simple: compute the key for each record, and write it only if that key is not already stored. Here the “table” is a dict keyed by hash:

import hashlib
import json

FACT_FIELDS = ("user_id", "event_type", "event_ts")

def key(rec):
  fact = {f: rec[f] for f in FACT_FIELDS}
  text = json.dumps(fact, sort_keys=True, separators=(",", ":"))
  return hashlib.sha256(text.encode("utf-8")).hexdigest()

table = {}   # stands in for a real table with a unique key column

def load(batch):
  new = 0
  for rec in batch:
      k = key(rec)
      if k not in table:               # a hash lookup, not a scan
          table[k] = rec
          new += 1
  return new

batch = [
  {"event_id": 48, "user_id": 3, "event_type": "app_open", "event_ts": "2025-03-04 10:51:00"},
  {"event_id": 49, "user_id": 3, "event_type": "app_open", "event_ts": "2025-03-04 10:51:00"},
  {"event_id": 50, "user_id": 3, "event_type": "search", "event_ts": "2025-03-04 11:00:00"},
]

print("first run, new rows: ", load(batch))
print("retry,     new rows: ", load(batch))
print("rows in table:       ", len(table))

The batch has a duplicate inside it (events 48 and 49), and then the whole batch is loaded again. The table ends with 2 rows either way. Run it 1 time or 100 times, same result. That is idempotency.

In a real database you get the same behaviour from a unique constraint on the hash column plus an upsert: INSERT ... ON CONFLICT (event_key) DO NOTHING in PostgreSQL and DuckDB, or MERGE in warehouses. The database does the if k not in table check for you, with an index instead of a dict.

Choosing which copy wins

When two copies have the same key, which one do you keep? The loader above keeps the first one it sees. Often that is right. But if copies can differ in fields outside the key, say a corrected fare, decide on purpose:

• First wins: DO NOTHING. Simple and safe for append-only facts like events.
• Last wins: DO UPDATE. Right when a later copy is a correction.
• Explicit order: sort by a field like ingested_at and keep the earliest or latest. That is the row_number() dedup from the SQL window functions lesson.

Ship it

Write load_events.py:

1. Stream content/datasets/ridelog/events.csv with the generator pipeline from the last lesson.
2. Compute a SHA-256 key per row from user_id, event_type and event_ts only, using canonical JSON.
3. Keep a dict of key to row, and skip keys you have already seen.
4. Run the load twice in the same script and print the row count after each run.

You have shipped it when both counts print 2,976, the same number the SQL query above returns.