Data Processing Algorithms Implementation

The reasoning

The naive version writes the output, then records the offset. A crash between them means the offset says less was done than actually was, so the batch is reprocessed — safe only if the output is idempotent. Reverse the order and a crash means the offset claims work that never happened, and a batch is silently skipped. Neither ordering is safe on its own; the ordering only decides which failure you get.

So make it one write. The example beside this writes offset and accumulated total into a single JSON file via write-to-temp-then-rename, which is atomic on POSIX — the file is either the old state or the new one, never a mixture. In a database it is one transaction containing both. In a streaming framework it is the checkpoint barrier, which is the same idea implemented for you.

Where a single atomic commit is impossible — output to object storage, offset in a database — you fall back to idempotence: make reprocessing a batch harmless, then choose the ordering that reprocesses rather than skips. That is the at-least-once-plus-idempotent shape again, and it is why the two ideas are inseparable.

The details that decide whether it works in practice. Batch size trades recovery time against commit overhead: small batches mean less rework and more commits. The state must be recoverable from durable storage, not held in memory. And the resume path must be the *ordinary* path — the example calls the same function with no special flag, which is what stops the recovery code from being the least-tested code in the system.

The reasoning

The principle is that the expensive part and the visible part must be separated. Write everything to a staging path or a temporary table while consumers continue reading the previous version, then flip in a single metadata operation that either happens or does not.

The mechanisms, by substrate. On a filesystem, write to a temp directory and rename — atomic within a filesystem, though *not* on object stores, where rename is a copy and is neither atomic nor cheap. On a warehouse, load into a staging table and swap with an atomic `ALTER TABLE ... RENAME` or a partition exchange. Behind a view, repoint the view at the new table. And with a table format like Iceberg or Delta, the commit is a single atomic metadata update by design, which is exactly the problem those formats were built to solve.

The object-store caveat deserves its own sentence because it catches people: writing a thousand objects to a prefix makes them visible one by one as they land, so a consumer listing that prefix mid-write sees a partial dataset. There is no rename to save you. That is why a table format's manifest — one atomic pointer to the set of files that constitute the current version — is the standard answer on object storage rather than a directory convention.

Two things to add. Keep the previous version until the new one is verified, so rollback is a second flip rather than a rebuild. And make the swap idempotent too: a retry of the publish step should be a no-op if it already happened, or you get two swaps and a confusing history.

The reasoning

The schedule: start at a base delay, double it each attempt, cap it so it does not grow to uselessness, and bound the total attempts. The measured table shows fixed backoff waiting five seconds over five retries and exponential waiting thirty-one — the point of exponential being that a dependency which is still down after a second is unlikely to be up a second later, so waiting longer costs nothing and offers more.

Jitter is the part people omit because, for one caller, it does nothing useful. Its effect is on the population: without it, every caller that failed at the same instant retries at the same instants forever after, so the service is hit by synchronised waves. The simulation beside this shows five hundred retries landing in the same half-second without jitter and a third fewer with it — the same retries, spread out.

The control that is most often missing is a retry *budget*: a limit on what fraction of total requests may be retries, typically enforced with a token bucket. Backoff bounds how fast one caller re-offers load; a budget bounds the amplification across all of them, which is the thing that turns a blip into an outage.

And the decision about what to retry at all. Transient failures — timeouts, 429, 503, connection resets — are worth retrying. Deterministic ones — 400, a validation error, a missing column — will fail identically and retrying is pure amplification that also delays the alert. Classify the error before retrying it, and pair the whole thing with a circuit breaker so that a dependency which is comprehensively down is left alone to recover.

The reasoning

**Closed** is normal: calls pass through and failures are counted, usually as a rate over a rolling window rather than a raw count, so a busy service is not tripped by a handful of errors. Past the threshold the breaker **opens**: every call fails immediately without touching the dependency. That fast failure is the point — the caller stops waiting on timeouts, and the dependency stops receiving traffic it cannot serve.

After a cooldown it moves to **half-open** and allows one request through. Success closes it; failure reopens it and restarts the cooldown. Without half-open you either stay open forever or reopen the floodgates blindly, so it is the state that makes recovery detectable at a cost of exactly one request.

What it adds over backoff: backoff makes each caller wait longer between attempts, but every caller still attempts, so a struggling service keeps receiving load from everyone. A breaker removes the load entirely for a period, which is what gives the dependency room to recover rather than being held down by the recovery attempts.

In a data pipeline the interesting question is what the open state *does*. Failing the task immediately is right when the dependency is essential. Skipping the enrichment and continuing with nulls is right when it is optional, and needs to be a deliberate, recorded degradation rather than a silent one. Buffering for later is right when the work can be deferred. Choosing among those is a business decision — and a breaker that opens with no defined behaviour is just a faster way to fail.

The reasoning

The implementation is two passes. First, route each event to `hash(key) % P` and append to that partition's file. Second, for each partition in turn, load its keys into a set and emit first occurrences. Only one partition's keys are resident at a time, so peak memory is set by `P` rather than by the input — the measured pair shows 12,600 keys held against the naive 198,654, with an identical answer.

Choosing `P`: distinct keys divided by what fits, with headroom — three to five times — because partitions are not equal. Skew matters here: a hash spreads keys evenly but not *rows*, and if one key has a billion occurrences its partition is huge even though its distinct count is one. Since deduplication only needs the distinct keys resident, that particular skew is harmless, which is a nice property worth pointing out.

If a partition's distinct keys still do not fit, recurse with a different hash seed. Using the same seed would put the same keys together again and make no progress, which is the classic bug in a hand-rolled implementation.

The details that decide whether it is usable. Which duplicate to keep needs a deterministic rule — latest by event time with a tie-break, not 'whichever came first', or the output changes between runs. The partition files are a spill to disk, so they need cleanup on failure. And the same structure parallelises for free: partitions are independent, so they can be processed on different machines, which is exactly what a distributed shuffle does.

The reasoning

The algorithm in three lines: fill the reservoir with the first k items; for each subsequent item at zero-based index `i`, draw `j = randrange(i + 1)`, and if `j < k` overwrite `kept[j]`. That is all of it.

Why it is uniform: the item at index `i` is kept with probability `k/(i+1)`, and an item already in the reservoir survives each subsequent step with probability `1 - 1/(i+1)`. Those telescope so that at the end every item has probability exactly `k/n`. The measured run bears it out — twenty items, a reservoir of five, two hundred thousand trials, and every item kept about fifty thousand times with a largest deviation under 1%.

The off-by-one is the bug to name because it is so easy to write: the range must be inclusive of the new item, `randrange(i + 1)` rather than `randrange(i)`. Getting it wrong makes early items over-represented, and the result still *looks* like a sample — which is why the measurement matters more here than the argument.

The properties that make it worth using: one pass, constant memory, no advance knowledge of the length, and no second pass to normalise. The variants worth knowing are weighted reservoir sampling, where items have unequal probabilities, and the fact that reservoirs merge — two reservoirs from two shards can be combined into one uniform sample of the union with appropriate weighting, which is what makes it usable distributed. And the distinction from hash-based sampling: this gives a uniform sample of *events*, where hashing a key gives a consistent sample of *keys*, and they answer different questions.

The reasoning

The operation you perform per candidate is 'is this better than the worst thing I am currently keeping'. With a min-heap of size k, the worst kept element is the root, so the check is `count > heap[0]` in constant time and the replacement is a single `heapreplace` in log k. With a max-heap the root is the best element, which you never need to look at, and finding the worst would be a linear scan.

The cost is `n log k` rather than `n log n` for a full sort, and memory is k rather than n. The measured run holds ten entries while a full sort would have ordered fifty thousand, and its result is asserted to equal the full sort exactly.

Ties are the part that makes a naive implementation non-deterministic. When the k-th and the (k+1)-th have the same count, which is kept depends on arrival order and on the heap's internal comparison. The measured run shows five keys tied on the boundary count. The fix is a total order: push `(count, key)` so ties break on the key, and compare the full sort the same way — `most_common` breaks ties by insertion order, which is not an ordering anyone should depend on.

Two refinements. Pushing the negated count into a min-heap is the standard trick for a max-heap in Python and is the opposite of what you want here, so it is worth being explicit about which you mean. And for the distributed case, k per node then merge is not exact — a key just below the cut on every node can beat one that was top on a single node — so keep more than k locally, which is the same caveat as the design question.

The reasoning

**Copy-on-write**: an update rewrites every data file containing an affected row. A few thousand scattered updates can touch a large fraction of the files, so the write cost is wildly disproportionate to the change — this is the write amplification that makes 'update one row' expensive on a columnar table. Readers, though, see clean files with no reconciliation, which is as fast as reads get.

**Merge-on-read**: the update is recorded as a delta file or a delete vector, so the write is proportional to the change rather than to the table. Every reader then merges base plus deltas, so read cost grows with unmerged changes since the last compaction — which is the compaction-debt problem, and it is why the mode requires a compaction schedule rather than merely benefiting from one.

For this workload — a few thousand updates against a billion rows, read many times a day — I would default to merge-on-read with scheduled compaction, because the write amplification of copy-on-write on scattered updates is the dominant cost and the read penalty between compactions is small. If the table were read rarely and updated in bulk, or if read latency were a hard requirement with no maintenance window, copy-on-write is the better fit.

The factor that often decides it in practice is clustering. If the updates are concentrated — always recent partitions — copy-on-write rewrites few files and its disadvantage largely disappears. If they are scattered uniformly across three years of history, it is the worst case. So the honest answer is to look at where the updates land before choosing, and to note that neither mode removes the work: they decide who pays and when.

The reasoning

**All-or-nothing** is right when the batch is semantically atomic — a financial posting, a set of rows that must be consistent. It requires the write to be rollback-able or idempotent so the retry can redo the whole batch, and it means one bad record blocks nine thousand four hundred good ones, which is head-of-line blocking at the batch level.

**Partial success** writes what it can and routes the failures elsewhere. It maximises throughput and it means the output is now incomplete in a way that must be recorded, or you have invented a silent data-loss mechanism. That record is the dead-letter or quarantine table: the record, the reason, the timestamp, enough to reprocess.

Whichever you choose, three things are non-negotiable. The failures must be persisted with their reason — a log line is not persistence. The failure *rate* must be monitored, because six hundred out of ten thousand is a systemic problem rather than six hundred bad records, and a threshold that fails the job is the difference between a data-quality signal and a slow leak. And there must be a defined path back: reprocessing the quarantine after a fix, rather than a table that only grows.

The judgment I would offer: partial success with quarantine as the default, all-or-nothing where atomicity is a real business requirement, and a failure-rate threshold in both cases. And the specific trap — retrying the whole batch when only some records failed will re-apply the successful ones, so partial success plus retry requires idempotent writes or per-record tracking, which is the interaction people miss.

The reasoning

**Parallelism.** Tasks finish in arbitrary order, so rows land in files in arbitrary order and file boundaries fall in different places. The content is the same set; the bytes are not. This is usually harmless and is the reason a naive byte comparison of two runs fails.

**Tie-breaks.** `row_number()` over an ordering with ties, a dedup keeping 'any' row per key, `first_value` over an unordered frame. These change the actual *content* between runs, which is a correctness problem rather than a cosmetic one, and it hides until the day someone notices a number moved.

**Runtime values.** `current_timestamp`, generated UUIDs, a run id embedded in the data. These make every run differ by construction, which defeats comparison — and if the timestamp is used as an updated_at, it also defeats any downstream change detection based on it.

**Iteration order.** Iterating a hash map and writing in that order, or floating-point sums combined in a different order producing a slightly different total.

Whether it matters depends on what consumes it. For reconciliation and testing it matters a great deal, and the fix is to make the *content* deterministic — total orderings on every tie-break, no wall-clock values in the data, a stable hash — and to compare content rather than bytes, with a sorted checksum. What is rarely worth chasing is byte-identical output, which requires pinning parallelism and file boundaries and buys almost nothing over a content comparison.

The reasoning

**Backpressure** propagates slowness upstream so the producer sends less. It preserves every record and it means the slowness becomes someone else's problem — which is right within a system you control and is not available when the producer is a customer's mobile app or a third party.

**Load shedding** drops data, deliberately and visibly: sample, drop a lower-priority stream, or reduce fidelity. It preserves latency for what remains and gives up completeness. The important word is deliberately — the alternative is dropping at random when a buffer overflows, which is the same loss with no control over what was lost.

**Buffering** absorbs a burst and does nothing for a sustained deficit. A queue in front of an under-provisioned consumer fills at the deficit rate until it hits its bound, and then you are shedding or backpressuring anyway — just with a large latency added first. Buffers are for smoothing spikes, and treating one as a solution to a rate mismatch is the most common mistake here.

So the first question is which situation you are in: a burst, where a bounded buffer sized to the burst is exactly right; or a deficit, where the rates must be reconciled by making the consumer faster, the work smaller, or the input less. And the design detail that decides whether the system degrades or collapses is that every buffer must be bounded — an unbounded queue converts a throughput problem into an out-of-memory failure, which is strictly worse because it also loses the buffer's contents.

The reasoning

Assign each key to one of P partitions by a hash, and keep a hash table per partition. When total memory hits the budget, choose a partition — usually the largest — and write its accumulators to disk, freeing that memory. Continue aggregating; keys belonging to a spilled partition are appended to its file rather than accumulated in memory. At the end, for each spilled partition, read its file, aggregate it in memory, and combine with whatever was already written.

Spilling whole partitions rather than individual keys is what makes the recovery cheap: because all of a key's rows are guaranteed to be in one partition, each spilled partition can be finished independently with no cross-partition merge. Evicting arbitrary keys would leave every key potentially split across memory and disk, and the final combine would be a full merge.

The partial-aggregation property is what makes it work at all: you can spill *accumulators*, not raw rows, so a partition that has seen a million rows for a thousand keys spills a thousand entries. That is why this is viable for sums and counts and awkward for medians, which are not decomposable — a median needs the values, so spilling is a sort rather than an aggregation.

Two details that bite. If a spilled partition is still too large to aggregate in memory, recurse with a *different hash seed* — reusing the seed puts the same keys together again and makes no progress. And the choice of which partition to evict matters: the largest frees the most memory, but if its keys keep arriving you spill it repeatedly, so a good implementation prefers a partition that is large *and* cold.

The reasoning

The key property is a *monotonic counter*, not a percentage. A percentage requires knowing the total, which you often do not, and it famously sticks at 99%. A counter that increases tells you work is happening; a counter that has not moved in ten minutes tells you it is not. That single distinction is what an operator actually needs at 3am.

Emit on a timer rather than per record, or the reporting itself becomes the bottleneck — every few seconds is plenty. Include enough to locate the work: records processed, the current partition or offset, elapsed time. The offset is the most useful field, because it also tells you where a restart would resume from.

Then make it visible somewhere durable — a metrics system, a status table, structured log lines — rather than only stdout, so it can be alerted on. The alert that matters is on the *derivative*: 'the counter has not increased in N minutes' catches a wedge, where a timeout on total duration catches it only at the end and cannot distinguish slow from stuck.

The related mechanism worth naming is a liveness heartbeat separate from progress: a process can be alive and making no progress — blocked on a lock, waiting on a dead connection — and those are different diagnoses. That is exactly what a distributed system's zombie detection is doing, and it is why a heartbeat is not the same thing as a progress counter even though both are timers.

The reasoning

Small N means frequent commits. Each commit has a fixed cost — a transaction, a file write, a metadata update — so at N=1 the overhead dominates and throughput collapses. The benefit is that a crash loses at most one record's work.

Large N means the opposite: overhead amortises to nothing, and a crash just before a commit discards everything since the last one. At N equal to the whole input you have a job with no checkpointing at all, which is the case where an eight-hour run fails at hour seven and starts again from zero. Large N also means more uncommitted state held in memory, which is its own limit.

So the sizing rule is expected-rework against overhead: choose N so that a commit costs a small fraction of the time to produce a batch — a few percent — and so that redoing one batch is an acceptable loss. In practice that usually lands at seconds-to-minutes of work per batch rather than at a particular record count, which is why *time-based* batching is often better than count-based: a fixed count means wildly different batch durations as record cost varies.

The same trade recurs everywhere and it is worth naming: streaming checkpoint intervals, file sizes on write, and shuffle partition counts are all this decision in different clothes. And the interaction with idempotence — if the write is idempotent, a slightly-too-large N costs only time on recovery; if it is not, batch boundaries become correctness boundaries and the pressure to get N right is much higher.

The reasoning

**Unit tests** on the logic, which requires the logic to be separable from the I/O. A function taking rows and returning rows can be tested with ten hand-written fixtures covering the edge cases — nulls, duplicates, boundary timestamps, the empty input. If the job is one monolithic script, this is a refactor before it is a test.

**Property tests** for the things that must hold regardless of input: running it twice produces the same result, processing a batch twice equals processing it once, the output row count relates to the input in a stated way, no key appears twice. Idempotence in particular is a property, not an example, and it is far better tested by generating inputs than by writing cases.

**Fixture-based integration tests** on a small realistic dataset through the real code path, checking the output matches an expected table. This is what catches wiring errors — the wrong column, a lost filter — that unit tests on individual functions miss.

**Production assertions**, which are the part people leave out and the only thing that covers scale and real data. Row counts reconciled against the source, uniqueness on the key, freshness, and volume within a band of the trailing average. A billion-row correctness property cannot be established in CI; it can be checked every night, and a failing check is worth more than a passing test suite.

What I would automate first: the idempotence property and the production reconciliation. Those two catch the failures that actually happen — duplicated data after a retry, and silent shortfalls — and neither is covered by the unit tests most teams write instead.

The reasoning

**Replace a partition.** The write deletes the rows for the interval it owns and inserts its own, in one transaction. Running twice gives the same result because the second run deletes what the first inserted. This is the most common and most robust option, and it requires the job to own a well-defined slice — which is what deriving the partition from the logical date gives you.

**Merge on a key.** `MERGE ... ON target.id = source.id WHEN MATCHED THEN UPDATE WHEN NOT MATCHED THEN INSERT`. Repeated application converges, provided the source itself has no duplicate keys within the batch — which it often does, so a deduplication step before the merge is usually part of the implementation rather than an optional extra.

**A constraint.** A primary key or unique index that rejects the duplicate outright, with the insert written to ignore conflicts. Simple and exact where the store enforces constraints; most analytical warehouses do not, which is why this is more common in operational stores than in the warehouse.

**An operation ledger.** Record a deterministic id for the unit of work — dag id, logical date, batch number — and skip if it is already present. The id must be derived, not generated, or a retry produces a new id and the check never fires. This is the fallback when the data has no natural key, and it needs the ledger write and the data write to be atomic, which returns to the checkpoint problem.

What none of them tolerate is a write that appends and a target with no key or partition. If the design has neither, that is the thing to fix — retrofitting idempotence onto an append-only write with no identity is not an implementation detail, it is a data model change.

The reasoning

A `try/finally` or a context manager handles the exception cases — an error, an interrupt — and handles none of the ones that matter most. `SIGKILL`, an OOM kill, a node termination and a power failure all stop the process without running any Python. So a design whose only cleanup is in the process is a design that leaks under exactly the conditions that produce the most garbage.

That makes cleanup an external responsibility. The standard approach is to write temp files under a path that identifies the run — `/tmp/job/<run_id>/` — and have a sweeper delete directories whose run is not active, or simply anything older than a threshold. Object stores give you this for free with a lifecycle policy on a temp prefix, which is the cheapest correct answer when the temp space is remote.

Which is also why the temp path should be deterministic: a rerun of the same unit of work should reuse or overwrite the same location rather than creating a new one, so a repeatedly-failing job does not multiply its garbage. Random temp names make every retry a fresh leak.

The related question is whether a partial temp file can be *mistaken* for a complete one, which is the more dangerous failure. If a restart might read what the previous attempt left, the files need a completion marker or an atomic rename on finish, so a partial file is never visible under its final name. That is the same publish-atomically principle applied one level down, and skipping it is how a resumed job produces silently truncated results.

The reasoning

**An added column** is additive and does not invalidate anything already produced. The options are to ignore it — safe, and it means nobody notices for six months — or to propagate it, with historical rows null for it. Both are defensible; what is not defensible is having no decision, so that whether it appears depends on whether the job used `select *`.

**A changed type** is a break. An int becoming a string may parse, coerce, or silently produce nulls depending on the engine, and any of those corrupts the column with no error. This should fail the job. A pipeline that quietly accepts type changes is one that will one day write nulls over a year of good data and report success.

**A removed column** is also breaking, and it fails at different times depending on the code: an explicit select fails immediately, which is good, and a `select *` succeeds and produces a narrower table, which is bad. That asymmetry is one of the better arguments for explicit column lists in anything that matters.

The mechanism that makes this systematic rather than ad hoc is a declared expected schema, checked at ingest. Compare what arrived against what is expected, classify the difference as additive or breaking, and act — accept and record additive changes, fail on breaking ones, and notify either way. That is what a data contract is, and it converts schema drift from something discovered downstream into something detected at the boundary. The practical version, if a contract is too much, is at minimum a schema check that fails on type changes, because that is the case that corrupts silently.

The reasoning

The row: the partition key — usually a date or a date-and-source — a state from `pending`, `running`, `succeeded`, `failed`, an attempt count, a last-updated timestamp, and ideally a summary of what was produced such as a row count. That last field is what makes reconciliation possible later without re-reading the output.

The transitions: a scheduler creates `pending` rows for units it expects to exist, claims one by moving it to `running` — with a conditional update so two workers cannot claim the same row — does the work, then marks `succeeded`. Failure increments attempts and returns it to `pending` or moves it to `failed` past a threshold.

The critical detail is that `succeeded` and the output must commit together, exactly as with the checkpoint. If the ledger is updated after the write, a crash between them leaves work done and unrecorded, so it will be redone — acceptable if idempotent. If it is updated first, a crash leaves work recorded and undone, which is a silent gap and is the failure a ledger was supposed to prevent. Where one transaction is impossible, order it so the failure is a redo rather than a skip, and rely on idempotence.

Then the properties that make it useful: 'what is missing' becomes `WHERE state != 'succeeded'`, backfill becomes setting rows to `pending`, and the same query drives monitoring. Add a bounded retention policy or the table grows forever, and add a stale-`running` sweeper — a row that has been `running` for longer than any real execution belongs to a dead worker and must be reclaimed, which is the same zombie-detection problem in a different costume.

The reasoning

**Idempotence first**, because everything else depends on it. Run it twice for the same input and diff the target. If the result differs, no retry is safe, no clear is safe, and no backfill is safe — and every one of those will happen. This is a ten-minute check that determines whether the job is operable at all.

**Resumability.** Kill it halfway and restart it. Does it resume, redo, or corrupt? A job that must run to completion or start over is one that cannot be run at all in an eight-hour window if it takes seven hours and the cluster is preemptible.

**Bounded memory.** Does its footprint depend on input size or on a fixed budget? A job holding a set that grows with distinct keys works for a year and dies the day a customer onboards. Look for unbounded accumulators, not for current usage.

**Failure behaviour.** Does it raise on error, or catch and continue? A swallowed exception makes a green run that did nothing, which is the worst outcome. Are there timeouts on external calls, retries with backoff, and a bound on how long it may run?

**Observability.** If it succeeded and produced nothing, would anyone know? That needs an output-level assertion — row counts reconciled, freshness checked — because task success is not evidence. And is there enough progress reporting to tell slow from stuck?

The one I would not compromise on is the first. A job that is not idempotent is one where every incident becomes a bespoke recovery, and the time to discover that is before it is scheduled rather than at 3am on a Sunday.