Streaming and Incremental Lineage Capture

Part of: Python Automation & Pipeline Integration

Batch lineage assumes a shape that a growing share of geospatial work no longer has. A nightly job reads a set of files, transforms them, writes an output, and emits a record describing the whole thing — clean, bounded, and easy to reason about. Sensor ingest, tile generation, change-only updates and streaming feeds break every part of that assumption. There is no single run to describe, the output is amended rather than created, and the volume of individual events is high enough that recording each one at batch granularity would produce more lineage than data.

Streaming and incremental lineage capture is the set of techniques for keeping provenance meaningful under those conditions. The core problem is not throughput, which is solvable with the buffering described in Asynchronous Logging Strategies. It is granularity: deciding what constitutes a step when work arrives continuously, and how a dataset that is perpetually being amended can have a derivation history that anyone can query.

In this guide

What Counts as a Step When There Is No Run

Batch provenance gets its natural unit for free: the job. Streaming work has no equivalent, so the unit has to be chosen — and choosing it badly produces either an unusable volume of records or a granularity too coarse to answer anything.

Three candidate units for a streaming step Per-message, per-window and per-checkpoint granularity compared on volume produced and the questions each can answer. UNIT VOLUME CAN ATTRIBUTE Per message one record per sensor reading unbounded everything — including things nobody will ever ask Per window ← usually right a time or count boundary predictable "which window produced this, from which source offsets" Per checkpoint whenever state is committed low "the job was running" — rarely enough on its own A window with recorded source offsets is reproducible; a checkpoint alone is not.

The window is the right default because it is the smallest unit that is independently reproducible. A record naming the window boundaries and the source offsets it consumed lets someone reconstruct exactly which inputs produced a given output — which is the question provenance exists to answer — without one record per reading. Per-message granularity buys attribution nobody needs at a cost that grows without limit, and per-checkpoint granularity omits the offsets, so the record says work happened without saying on what.

Choose the window by what a consumer would consider one logical update. For a tile pipeline that is usually a tile and a zoom level; for a sensor feed, a time bucket aligned to how the data is published; for a change feed, the batch of changes applied together. Aligning the lineage window to the publication unit means each published artefact has exactly one producing record, which keeps the graph interpretable.

Incremental Updates and the Amended-Dataset Problem

Batch lineage models a derivation as one output produced from some inputs. Incremental work breaks this because the output already existed and was modified — and a graph that records “dataset X was derived from batch 47” overwrites the fact that it was also derived from batches 1 through 46.

Versions, not overwrites, for incremental updates Overwriting a dataset's derivation edge each update leaves only the most recent input; versioning with supersedes edges keeps the whole history. OVERWRITE — history lost parcels (one node) derived_from: batch 47 Batches 1–46 are no longer reachable from the dataset "which update introduced this error?" — unanswerable VERSION — history preserved parcels@v46 from batch 46 supersedes parcels@v47 from batch 47 Each version names its own input. The chain of supersedes edges is the history. A version node per update costs one row and makes every past state addressable. The logical dataset identity is a separate, stable attribute shared by every version.

Versioning is the same conclusion the relational side reaches in Versioning Lineage Rows with Temporal Tables, arrived at from a different direction. Keeping the logical identity stable while giving each state its own node means a downstream product can reference the exact version it consumed, which is the only way an analysis run last March remains explicable after forty subsequent updates.

The cost worth managing is version count. A feed producing an update every minute generates half a million versions a year, and while that is storable it makes the supersedes chain unpleasant to walk. Where updates are frequent and individually insignificant, aggregate them into a published version at the cadence consumers actually observe, and keep the individual updates as attributes of that version rather than as nodes.

Event Volume and the Aggregation Boundary

The decision that determines whether a streaming lineage design is workable is where aggregation happens. Aggregating too early loses the detail that makes records useful; too late means moving and storing volume nobody needs.

Aggregate at the point where individual events stop being independently interesting. For a tile pipeline, that is usually the tile: knowing which source features contributed to tile 12/2045/1362 is useful, while knowing the order in which its pixels were rasterised is not. For a sensor feed it is the time bucket, because a question about a single reading almost always turns out to be a question about the period containing it.

Record aggregate statistics rather than discarding the detail silently. A window record carrying the message count, the source offset range, the count of messages rejected and the distribution of event timestamps preserves most of the diagnostic value of the individual events at a fraction of the volume. What it cannot answer — “what happened to this specific reading” — is worth being explicit about, since a lineage system whose limits are documented is one people can rely on.

Where per-event provenance is genuinely required, and occasionally it is, keep it out of the lineage store. Write it alongside the data itself as an attribute or a sidecar, referenced from the window record. That keeps the queryable graph proportional to windows while leaving the detail recoverable, which is the same header-and-body split described in Structuring JSON/XML Lineage Documents.

Where the detail goes when events are aggregated Individual events collapse into one window record carrying summary statistics; per-event detail is written beside the data and referenced, not stored in the graph. EVENTS IN … thousands more, one of them rejected One window record — in the graph offsets 40912–44108 · accepted 3,195 rejected 1 · event-time span, processing time detail_ref → sidecar digest Sidecar — beside the data per-event rows, if anyone needs them content-addressed, never scanned cheap storage, no index The graph stays proportional to windows; the detail stays recoverable. Record the reject count in the window — a silent drop is the failure this design most needs to expose.

The reject count in the window record deserves emphasis because it is the difference between an aggregate that hides problems and one that surfaces them. A window reporting three thousand accepted messages tells you throughput; the same window reporting one rejection tells you the feed contains something the parser did not expect, which is usually the first sign of an upstream schema change. Aggregation is only safe when the aggregate carries the anomalies rather than averaging them away.

Choose the sidecar’s granularity to match how anyone would actually look at it. A file per window, content-addressed and written to object storage, is queryable enough for a human investigating one window and cheap enough to keep indefinitely. Writing per-event rows into a database because it feels more accessible reintroduces the volume problem the aggregation was meant to solve.

Ordering, Duplicates and Late Arrivals

Streaming introduces three conditions that batch pipelines rarely face, and each affects what a lineage record can honestly claim.

Out-of-order arrival means the window a message belongs to is not the window it arrived in. Record both the event time and the processing time on every window record, because a consumer asking “what did we know at 3pm” needs processing time while one asking “what happened at 3pm” needs event time. Systems that record only one answer the wrong question half the time.

Duplicates are normal under at-least-once delivery and must not produce duplicate lineage. The content-derived key described elsewhere on this site handles it: a redelivered message produces an identical record that upserts harmlessly. What needs care is the window aggregate, since a duplicate message inflates a count that was computed before deduplication — deduplicate first, then aggregate.

Late arrivals are the hardest, because a window may already have been closed and its record written. Reopening it violates append-only; ignoring the message loses data. The workable answer is a correction record: a new event superseding the window’s earlier aggregate, with an explicit reason. That keeps the history honest — the original figure was what was known at the time — and makes the correction itself auditable, which is exactly the property that Regulatory Compliance & Standards Mapping requires of any amendment.

Set a lateness horizon and record it. After some interval a window is final and later messages are routed to a dead-letter path rather than triggering endless corrections. Stating that interval in the record means a consumer knows how much to trust a recent window versus an old one.

Tiles Are the Common Case

Most geospatial teams meet incremental lineage first through tile generation, and it is worth working through because it exercises every idea above at once.

A tile pyramid is an output that is perpetually partially rebuilt. A change to a single parcel invalidates a handful of tiles at high zoom and one tile at low zoom, so a regeneration touches a scattered subset rather than the whole set. Modelling the pyramid as one dataset with one derivation edge is therefore useless — the edge would have to name every source feature that has ever contributed, and it would be rewritten on every update.

The workable model treats each tile as its own entity, with the pyramid as a logical collection rather than a node. A regeneration produces a version of one tile, derived from the source features intersecting its extent at that moment, recorded once per tile rather than once per feature. That gives an answer to the question tile pipelines actually attract — “why does this tile show the old boundary?” — which resolves to reading one tile’s derivation record and comparing its source version against the current one.

Volume is manageable because tiles are already a bounded set. A pyramid to zoom 14 over a county is tens of thousands of tiles, and a version record per regenerated tile is proportional to how much actually changed rather than to the pyramid’s size. Where a full rebuild happens, record it as one bulk activity producing many tile versions rather than as tens of thousands of independent steps, which keeps the graph honest about the fact that they were one operation.

The zoom hierarchy adds one wrinkle worth handling explicitly: a low-zoom tile is derived from high-zoom tiles in some pipelines and from source data in others. Record which, because the two produce different answers to an impact query and the difference is invisible from the output.

Configuration Reference

Parameter Type Valid values Default
window_kind enum time, count, session, publication_unit publication_unit
window_size duration or int Aligned to how consumers observe the data none (required)
lateness_horizon duration After which a window is final and late events dead-letter 1h
record_event_and_processing_time boolean Both, always — they answer different questions true
dedup_before_aggregate boolean Otherwise duplicates inflate window counts true
version_on_update boolean Create a version node per published update true
per_event_detail enum none, sidecar, inline — inline only for low volume sidecar
offset_range_recorded boolean Source offsets are what make a window reproducible true

offset_range_recorded is the field that distinguishes a reproducible window record from a descriptive one. Without the source offsets, a record says a window ran and produced output; with them, someone can replay exactly the same input range and check the result. For Kafka-style sources this is a partition-and-offset range; for a file feed it is a set of file identifiers with digests; for a database change feed it is a log sequence range.

lateness_horizon deserves an explicit value rather than an implicit infinity. Correction records are legitimate and unbounded correction is not — a system still amending windows from two years ago has no stable history for anyone to cite. Pick a horizon from how late data actually arrives, measure the distribution, and revisit it rather than guessing once.

Common Failure Modes and Mitigations

Failure mode Symptom Mitigation
Window record without offsets Cannot reproduce or verify what a window consumed Require an offset or file-digest range on every window record
Duplicate inflation Counts higher than the true message volume Deduplicate on the content key before computing aggregates
Overwritten dataset node Only the most recent update is reachable Version nodes with supersedes edges; stable logical identity as an attribute
Event time and processing time conflated “What did we know then” and “what happened then” give the same wrong answer Record both; never derive one from the other
Unbounded corrections Old windows keep changing; no citable history Set and record a lateness horizon; dead-letter beyond it
Per-event records in the lineage store Graph grows faster than the data it describes Aggregate at the publication unit; put per-event detail in a sidecar

The version-overwrite row is the one that causes the most damage per occurrence, because it is silent and cumulative. Each update quietly removes the previous derivation edge, so the graph always looks complete for the current state and has no memory at all. It is typically discovered when someone asks which update introduced a defect and finds that the question has no representation in the data.

Compliance and Governance Alignment

Control / framework Requirement What streaming capture supplies
ISO 19115 LI_ProcessStep Each processing step described One step per window, with its source range and parameters
GDPR Article 30 Records of processing activities Windows carry purpose and lawful basis like any other activity
FISMA AU-2 Auditable events defined and captured Window records plus dead-letter events for what was refused
Retention schedules Disposal on a documented timetable Version nodes age out per schedule; the supersedes chain records the disposal
ISO 19157 Data quality reported Per-window reject counts and completeness ratios as quality metrics

The retention row raises a consideration specific to versioned streaming data. A dataset with half a million versions has half a million retention decisions unless versions are aggregated, and applying a schedule to each individually is impractical. Define retention over published versions rather than over internal update states, and let the internal states expire on a shorter clock — recording that policy as an explicit rule rather than letting it emerge from whatever the storage layer does.

Frequently Asked Questions

Does streaming lineage need a streaming platform?

No. The techniques here apply equally to a job that runs every five minutes over new files, which is how much “streaming” geospatial work is actually implemented. What matters is the granularity decision and the versioning model, not whether a message broker is involved.

How do we handle a window that produced nothing?

Record it. An empty window is a fact — the feed was healthy and had nothing to report — and it is distinguishable from a window that never ran only if it left a record. Gaps in the window sequence are one of the clearest signals that ingestion stopped, and they are visible only if the normal case writes something.

What identifier should a window record use?

Derive it from the window boundaries and the source, so it is stable across retries and reproducible from the inputs. A UUID works and loses the property that two workers processing the same window independently produce the same identifier, which is what makes deduplication free.

Can we reconstruct per-feature lineage from window records?

Only to window granularity, which is usually the honest answer to give. If a consumer needs to know which source reading produced a specific output feature, that requires per-event capture — record it in the data rather than in the lineage graph, and reference it from the window.

How does this interact with backfills?

A backfill is a set of windows processed out of chronological order, which is exactly why event time and processing time must both be recorded. Route backfill output to its own partition where the storage layer benefits from it, as described in Spatial Partitioning for Lineage Tables, and mark the records so a consumer can distinguish reconstructed history from contemporaneous capture.

What happens to lineage when the stream is replayed?

Nothing, if the design is right. Replaying a window produces a record with the same content-derived key and the same content, which upserts to a no-op. That property is worth testing deliberately, because it is also what protects against duplicate delivery, and a system where replay creates new records will accumulate them invisibly.