Read every event once. Survive every crash. Never count anything twice.

Real-Time & Streaming — Pipelines That Survive Crashes¶

Think of a little train that carries boxes. Each box is one taxi ride.

Sometimes the train crashes. When it starts again, two bad things used to happen:

• 😱 It forgot some boxes — you lose taxi rides.
• 😱 It read some boxes twice — you charge people twice.

The fix is a well-known trick — it's how big streaming systems (like Spark and Kafka) already stay safe:

The train leaves a bookmark every time it safely drops off boxes. If it crashes, it wakes up, reads the bookmark, and starts again from that exact spot.

That idea isn't new. What LakeLogic adds is that you get it everywhere, from one rule book — the same crash-safe behavior whether you're on a laptop with Polars, DuckDB, or the big Spark engine. You don't rebuild it per tool.

The result: no lost rides, no double rides — every event is handled exactly once.

This whole notebook runs with no Kafka, no cloud, and no cluster — the data sources are tiny pretend versions so every cell just works here. In real life you swap the pretend source for a real one (real brokers, a real web feed, a real Spark stream); the crash-safe logic stays exactly the same.

The story: RideFlow¶

RideFlow is a taxi app. We'll bring its live data in safely, six ways:

#	What we want	What makes it happen
1	Read live taxi rides and survive a crash — lose nothing, double nothing	StreamSink + KafkaOffsetSource
2	Follow a live price feed and pick up right where we left off	SSEOffsetSource
3	Load a giant pile of old rides without running out of memory	WatermarkChunkSource
4	Do heavier math on PySpark (the big engine)	SparkStreamSink
5	Catch a bad plan before it makes double rides	streaming checks (STREAM-001/002)
6	Prove we never double-count, even after a replay	merge + replay

In [ ]:

# Install LakeLogic (Polars + DuckDB engines are all this notebook needs)
!pip install -q lakelogic[polars,duckdb]

import lakelogic as ll
from lakelogic import (
    StreamSink,
    SQLiteCheckpointStore,
    KafkaOffsetSource,
    SSEOffsetSource,
    WatermarkChunkSource,
    SparkStreamSink,
)

print("LakeLogic", ll.__version__)

# Install LakeLogic (Polars + DuckDB engines are all this notebook needs) !pip install -q lakelogic[polars,duckdb] import lakelogic as ll from lakelogic import ( StreamSink, SQLiteCheckpointStore, KafkaOffsetSource, SSEOffsetSource, WatermarkChunkSource, SparkStreamSink, ) print("LakeLogic", ll.__version__)

0. One rule book for every box¶

A contract is just a rule book for the data. It says what a good taxi ride looks like. The same rule book checks every box the train brings in — a few boxes at a time.

Our rule book keeps rides where the fare is not negative. Bad ones are put in a time-out box (kept, not thrown away). And strategy: merge on trip_id is the magic that stops double-counting later (see Section 6).

In [ ]:

TRIP_CONTRACT = {
    "version": "1.0.0",
    "dataset": "bronze_trip_events",
    "info": {"title": "bronze_trip_events"},
    "primary_key": ["trip_id"],
    "model": {
        "fields": [
            {"name": "trip_id", "type": "integer", "required": True},
            {"name": "rider_id", "type": "integer"},
            {"name": "fare", "type": "float"},
            {"name": "surge", "type": "float"},
        ]
    },
    "quality": {
        "row_rules": [
            {"name": "fare_positive", "sql": "fare >= 0"},
        ]
    },
    "materialization": {"strategy": "merge", "format": "delta", "target_path": "lake/bronze_trips"},
}


def trip_events(n, start=0, bad_every=25):
    """Simulate n RideFlow trip events; every bad_every-th has a negative (invalid) fare."""
    out = []
    for i in range(start, start + n):
        fare = -1.0 if (bad_every and i % bad_every == 0) else round(5 + (i % 40) * 1.5, 2)
        out.append({"trip_id": i, "rider_id": 1000 + (i % 500), "fare": fare, "surge": 1.0 + (i % 5) * 0.2})
    return out


print("sample event:", trip_events(1)[0])

TRIP_CONTRACT = { "version": "1.0.0", "dataset": "bronze_trip_events", "info": {"title": "bronze_trip_events"}, "primary_key": ["trip_id"], "model": { "fields": [ {"name": "trip_id", "type": "integer", "required": True}, {"name": "rider_id", "type": "integer"}, {"name": "fare", "type": "float"}, {"name": "surge", "type": "float"}, ] }, "quality": { "row_rules": [ {"name": "fare_positive", "sql": "fare >= 0"}, ] }, "materialization": {"strategy": "merge", "format": "delta", "target_path": "lake/bronze_trips"}, } def trip_events(n, start=0, bad_every=25): """Simulate n RideFlow trip events; every bad_every-th has a negative (invalid) fare.""" out = [] for i in range(start, start + n): fare = -1.0 if (bad_every and i % bad_every == 0) else round(5 + (i % 40) * 1.5, 2) out.append({"trip_id": i, "rider_id": 1000 + (i % 500), "fare": fare, "surge": 1.0 + (i % 5) * 0.2}) return out print("sample event:", trip_events(1)[0])

1. Live taxi rides — crash, then start again with nothing lost¶

What we want: RideFlow sends a stream of live rides. If our job dies (a restart, a crash, out of memory), it must start again from the exact spot it last saved — lose no rides, bill no one twice.

KafkaOffsetSource saves the bookmark after the boxes are safely written down. If it wakes up later, it jumps straight back to the bookmark.

Here we use a tiny pretend train so it runs anywhere. In real life you just write KafkaOffsetSource('trips', brokers='broker:9092', group_id='rideflow-bronze') — nothing else changes.

In [ ]:

# --- A stand-in broker so this notebook runs anywhere (production: pass real brokers) ---
from lakelogic.core.stream_sink import _SimpleTP


class _Rec:
    __slots__ = ("topic", "partition", "offset", "value")

    def __init__(self, t, p, o, v):
        self.topic, self.partition, self.offset, self.value = t, p, o, v


class InMemoryBroker:
    """A minimal kafka-python-shaped consumer over in-memory partitions."""

    def __init__(self, topic, by_partition):
        self.topic = topic
        self._data = {p: list(v) for p, v in by_partition.items()}
        self._pos = {p: 0 for p in self._data}
        self._assigned = []

    def partitions_for_topic(self, t):
        return set(self._data)

    def assign(self, tps):
        self._assigned = list(tps)

    def seek(self, tp, off):
        self._pos[tp.partition] = off

    def poll(self, timeout_ms=None, max_records=500):
        out = {}
        for tp in self._assigned:
            p, vals, start = tp.partition, self._data[tp.partition], self._pos[tp.partition]
            if start >= len(vals):
                continue
            end = min(len(vals), start + max_records)
            out[tp] = [_Rec(self.topic, p, o, vals[o]) for o in range(start, end)]
            self._pos[p] = end
        return out

    def close(self):
        pass


def kafka_source(by_partition):
    broker = InMemoryBroker("trips", by_partition)
    return KafkaOffsetSource("trips", consumer=broker, tp_factory=_SimpleTP, max_poll_records=200)

# --- A stand-in broker so this notebook runs anywhere (production: pass real brokers) --- from lakelogic.core.stream_sink import _SimpleTP class _Rec: __slots__ = ("topic", "partition", "offset", "value") def __init__(self, t, p, o, v): self.topic, self.partition, self.offset, self.value = t, p, o, v class InMemoryBroker: """A minimal kafka-python-shaped consumer over in-memory partitions.""" def __init__(self, topic, by_partition): self.topic = topic self._data = {p: list(v) for p, v in by_partition.items()} self._pos = {p: 0 for p in self._data} self._assigned = [] def partitions_for_topic(self, t): return set(self._data) def assign(self, tps): self._assigned = list(tps) def seek(self, tp, off): self._pos[tp.partition] = off def poll(self, timeout_ms=None, max_records=500): out = {} for tp in self._assigned: p, vals, start = tp.partition, self._data[tp.partition], self._pos[tp.partition] if start >= len(vals): continue end = min(len(vals), start + max_records) out[tp] = [_Rec(self.topic, p, o, vals[o]) for o in range(start, end)] self._pos[p] = end return out def close(self): pass def kafka_source(by_partition): broker = InMemoryBroker("trips", by_partition) return KafkaOffsetSource("trips", consumer=broker, tp_factory=_SimpleTP, max_poll_records=200)

In [ ]:

# Two partitions of live trips. Drain them through the contract, committing offsets.
ckpt = SQLiteCheckpointStore("checkpoints.sqlite")

src = kafka_source({0: trip_events(600, start=0), 1: trip_events(400, start=600)})
sink = StreamSink(
    TRIP_CONTRACT,
    src,
    engine="polars",
    checkpoint=ckpt,
    checkpoint_key="trips",
    batch_size=250,
    target_path="lake/bronze_trips",
)
summary = sink.run("available_now")  # AvailableNow: drain to current end, then exit

print(f"batches      : {summary.batches}")
print(f"events read  : {summary.source_count}")
print(f"valid  (good): {summary.good_count}")
print(f"quarantined  : {summary.bad_count}")
print(f"committed offset (cursor): {summary.cursor}")  # per-partition broker offsets

# Two partitions of live trips. Drain them through the contract, committing offsets. ckpt = SQLiteCheckpointStore("checkpoints.sqlite") src = kafka_source({0: trip_events(600, start=0), 1: trip_events(400, start=600)}) sink = StreamSink( TRIP_CONTRACT, src, engine="polars", checkpoint=ckpt, checkpoint_key="trips", batch_size=250, target_path="lake/bronze_trips", ) summary = sink.run("available_now") # AvailableNow: drain to current end, then exit print(f"batches : {summary.batches}") print(f"events read : {summary.source_count}") print(f"valid (good): {summary.good_count}") print(f"quarantined : {summary.bad_count}") print(f"committed offset (cursor): {summary.cursor}") # per-partition broker offsets

The bookmark is a real spot in the stream ({'trips:0': 600, 'trips:1': 400}), not a guess. Now let's pretend the train crashed, and show it starts again in exactly the right place — reading only the new boxes.

In [ ]:

# The topic has grown: 200 more events arrived on partition 0 since we last ran.
src2 = kafka_source({0: trip_events(800, start=0), 1: trip_events(400, start=600)})
sink2 = StreamSink(
    TRIP_CONTRACT,
    src2,
    engine="polars",
    checkpoint=ckpt,
    checkpoint_key="trips",
    batch_size=250,
    target_path="lake/bronze_trips",
)
resumed = sink2.run("available_now")

print(f"resumed from : {resumed.resumed_from}")  # seeks to the committed offsets
print(f"events read  : {resumed.source_count}")  # ONLY the 200 new ones — no reprocessing
print(f"new cursor   : {resumed.cursor}")

# The topic has grown: 200 more events arrived on partition 0 since we last ran. src2 = kafka_source({0: trip_events(800, start=0), 1: trip_events(400, start=600)}) sink2 = StreamSink( TRIP_CONTRACT, src2, engine="polars", checkpoint=ckpt, checkpoint_key="trips", batch_size=250, target_path="lake/bronze_trips", ) resumed = sink2.run("available_now") print(f"resumed from : {resumed.resumed_from}") # seeks to the committed offsets print(f"events read : {resumed.source_count}") # ONLY the 200 new ones — no reprocessing print(f"new cursor : {resumed.cursor}")

2. A live price feed — pick up right where we left off¶

What we want: RideFlow also sends a live feed of price changes. This kind of feed has its own built-in bookmark called Last-Event-ID, and LakeLogic knows how to use it.

The Kafka bookmark was a little list. This bookmark is just one number/name for the last thing we saw. Same idea, different shape — so the crash-safe trick isn't only for one kind of feed.

We use a pretend feed here. In real life you pass SSEOffsetSource('https://rideflow/surge/stream') and it picks up from the saved bookmark all by itself.

In [ ]:

def surge_feed(events):
    """Fake SSE server: replays every event AFTER the armed Last-Event-ID."""
    import json

    class Evt:
        __slots__ = ("id", "data")

        def __init__(self, i, d):
            self.id, self.data = i, d

    def connect(last_event_id):
        started = last_event_id is None
        for eid, payload in events:
            if not started:
                if str(eid) == str(last_event_id):
                    started = True
                continue
            yield Evt(str(eid), json.dumps(payload))

    return connect


price_events = [
    (i, {"trip_id": i, "rider_id": 1000 + i, "fare": 12.0, "surge": 1.0 + (i % 6) * 0.3}) for i in range(300)
]

sse_ckpt = SQLiteCheckpointStore("checkpoints.sqlite")
sse = SSEOffsetSource(connect=surge_feed(price_events))
run1 = StreamSink(
    TRIP_CONTRACT,
    sse,
    engine="polars",
    checkpoint=sse_ckpt,
    checkpoint_key="surge",
    batch_size=100,
    target_path="lake/bronze_trips",
).run("available_now")
print("drained events:", run1.source_count, "| Last-Event-ID committed:", run1.cursor)

def surge_feed(events): """Fake SSE server: replays every event AFTER the armed Last-Event-ID.""" import json class Evt: __slots__ = ("id", "data") def __init__(self, i, d): self.id, self.data = i, d def connect(last_event_id): started = last_event_id is None for eid, payload in events: if not started: if str(eid) == str(last_event_id): started = True continue yield Evt(str(eid), json.dumps(payload)) return connect price_events = [ (i, {"trip_id": i, "rider_id": 1000 + i, "fare": 12.0, "surge": 1.0 + (i % 6) * 0.3}) for i in range(300) ] sse_ckpt = SQLiteCheckpointStore("checkpoints.sqlite") sse = SSEOffsetSource(connect=surge_feed(price_events)) run1 = StreamSink( TRIP_CONTRACT, sse, engine="polars", checkpoint=sse_ckpt, checkpoint_key="surge", batch_size=100, target_path="lake/bronze_trips", ).run("available_now") print("drained events:", run1.source_count, "| Last-Event-ID committed:", run1.cursor)

In [ ]:

# Feed grows to 480 ticks; reconnect resumes from the committed Last-Event-ID.
price_events_grown = [
    (i, {"trip_id": i, "rider_id": 1000 + i, "fare": 12.0, "surge": 1.0 + (i % 6) * 0.3}) for i in range(480)
]
sse2 = SSEOffsetSource(connect=surge_feed(price_events_grown))
run2 = StreamSink(
    TRIP_CONTRACT,
    sse2,
    engine="polars",
    checkpoint=sse_ckpt,
    checkpoint_key="surge",
    batch_size=100,
    target_path="lake/bronze_trips",
).run("available_now")
print("resumed from Last-Event-ID:", run2.resumed_from)
print("only new ticks read       :", run2.source_count, "(events 300..479)")

# Feed grows to 480 ticks; reconnect resumes from the committed Last-Event-ID. price_events_grown = [ (i, {"trip_id": i, "rider_id": 1000 + i, "fare": 12.0, "surge": 1.0 + (i % 6) * 0.3}) for i in range(480) ] sse2 = SSEOffsetSource(connect=surge_feed(price_events_grown)) run2 = StreamSink( TRIP_CONTRACT, sse2, engine="polars", checkpoint=sse_ckpt, checkpoint_key="surge", batch_size=100, target_path="lake/bronze_trips", ).run("available_now") print("resumed from Last-Event-ID:", run2.resumed_from) print("only new ticks read :", run2.source_count, "(events 300..479)")

3. Load a giant pile of old rides — without running out of memory¶

What we want: RideFlow has years of old rides to load. If we try to grab them all at once, we run out of memory. And if it fails near the end, we don't want to start all over.

WatermarkChunkSource grabs the pile in small handfuls instead of all at once. Memory stays small (one handful at a time), and if it crashes it starts from the last handful — not from zero. A big batch job that gets the crash-safe trick for free.

You tell it how to grab one handful (fetch_chunk), and that can be any database or file — LakeLogic doesn't care which.

In [ ]:

# Simulate a 5,000-row historical table behind a keyset query.
history = trip_events(5000, start=0, bad_every=0)  # all valid, for a clean count
history_sorted = sorted(history, key=lambda r: r["trip_id"])


def fetch_chunk(after, limit):
    if after is None:
        start = 0
    else:
        start = next((i for i, r in enumerate(history_sorted) if r["trip_id"] > after), len(history_sorted))
    return history_sorted[start : start + limit]


backfill_ckpt = SQLiteCheckpointStore("checkpoints.sqlite")
backfill = WatermarkChunkSource(fetch_chunk, watermark_field="trip_id", chunk_size=500)
b = StreamSink(
    TRIP_CONTRACT,
    backfill,
    engine="polars",
    checkpoint=backfill_ckpt,
    checkpoint_key="backfill",
    batch_size=500,
    target_path="lake/bronze_trips",
).run("available_now")
print(f"rows loaded : {b.source_count}  (in {b.batches} bounded chunks of 500 — flat memory)")
print(f"watermark   : {b.cursor}  (last trip_id; resume continues strictly after this)")

# Simulate a 5,000-row historical table behind a keyset query. history = trip_events(5000, start=0, bad_every=0) # all valid, for a clean count history_sorted = sorted(history, key=lambda r: r["trip_id"]) def fetch_chunk(after, limit): if after is None: start = 0 else: start = next((i for i, r in enumerate(history_sorted) if r["trip_id"] > after), len(history_sorted)) return history_sorted[start : start + limit] backfill_ckpt = SQLiteCheckpointStore("checkpoints.sqlite") backfill = WatermarkChunkSource(fetch_chunk, watermark_field="trip_id", chunk_size=500) b = StreamSink( TRIP_CONTRACT, backfill, engine="polars", checkpoint=backfill_ckpt, checkpoint_key="backfill", batch_size=500, target_path="lake/bronze_trips", ).run("available_now") print(f"rows loaded : {b.source_count} (in {b.batches} bounded chunks of 500 — flat memory)") print(f"watermark : {b.cursor} (last trip_id; resume continues strictly after this)")

4. Bigger math on the big engine (PySpark)¶

What we want: Some math is heavy — like adding up prices over time windows. That belongs on the big, strong engine (Spark). The good news: we use the same rule book there too. LakeLogic just hops inside Spark and checks every handful of boxes with the same rules. Spark keeps the bookmark this time.

On any PySpark platform (Databricks, EMR, Dataproc, Synapse, or your own Spark cluster) it looks like this:

silver_events = (spark.readStream.format('delta').load('lake/bronze_trips'))

sink = SparkStreamSink(
    SILVER_SURGE_CONTRACT,
    silver_events,
    checkpoint_location='/checkpoints/silver_surge',  # Spark keeps the bookmark
    trigger='available_now',        # or run all the time: trigger='processing_time', processing_time='30 seconds'
    on_batch=lambda b: print(f'batch {b.batch_id}: good={b.good_count} bad={b.bad_count}'),
)
query = sink.run()

We don't have the big engine here, so we run the same handler with a pretend feeder — to show the wiring is real and each handful really goes through the rule book:

In [ ]:

# Prove the foreachBatch handler runs the contract per micro-batch (no Spark needed).
# The ONLY fake here is the writeStream wiring; the micro-batches are real frames
# and each one flows through the real contract engine. (We use engine='polars' so
# it runs locally; in production SparkStreamSink defaults to engine='spark'.)
import polars as pl


class FakeWriter:
    def __init__(self, batches):
        self._b, self.opts, self.trig = batches, {}, None

    def foreachBatch(self, fn):
        self._fn = fn
        return self

    def option(self, k, v):
        self.opts[k] = v
        return self

    def outputMode(self, m):
        return self

    def trigger(self, **kw):
        self.trig = kw
        return self

    def start(self):
        for i, frame in enumerate(self._b):
            self._fn(frame, i)  # Spark drives foreachBatch
        return self

    def awaitTermination(self):
        pass


class FakeStreamDF:
    def __init__(self, batches):
        self.writeStream = FakeWriter(batches)


micro_batches = [pl.DataFrame(trip_events(200, start=0)), pl.DataFrame(trip_events(200, start=200))]
df = FakeStreamDF(micro_batches)
spark_sink = SparkStreamSink(
    TRIP_CONTRACT,
    df,
    engine="polars",
    checkpoint_location="/checkpoints/silver_surge",
    target_path="lake/silver_surge",
    on_batch=lambda b: print(f"  micro-batch {b.batch_id}: good={b.good_count} bad={b.bad_count}"),
)
spark_sink.run()
print("checkpointLocation (Spark owns resume):", df.writeStream.opts["checkpointLocation"])
print("trigger:", df.writeStream.trig)

# Prove the foreachBatch handler runs the contract per micro-batch (no Spark needed). # The ONLY fake here is the writeStream wiring; the micro-batches are real frames # and each one flows through the real contract engine. (We use engine='polars' so # it runs locally; in production SparkStreamSink defaults to engine='spark'.) import polars as pl class FakeWriter: def __init__(self, batches): self._b, self.opts, self.trig = batches, {}, None def foreachBatch(self, fn): self._fn = fn return self def option(self, k, v): self.opts[k] = v return self def outputMode(self, m): return self def trigger(self, **kw): self.trig = kw return self def start(self): for i, frame in enumerate(self._b): self._fn(frame, i) # Spark drives foreachBatch return self def awaitTermination(self): pass class FakeStreamDF: def __init__(self, batches): self.writeStream = FakeWriter(batches) micro_batches = [pl.DataFrame(trip_events(200, start=0)), pl.DataFrame(trip_events(200, start=200))] df = FakeStreamDF(micro_batches) spark_sink = SparkStreamSink( TRIP_CONTRACT, df, engine="polars", checkpoint_location="/checkpoints/silver_surge", target_path="lake/silver_surge", on_batch=lambda b: print(f" micro-batch {b.batch_id}: good={b.good_count} bad={b.bad_count}"), ) spark_sink.run() print("checkpointLocation (Spark owns resume):", df.writeStream.opts["checkpointLocation"]) print("trigger:", df.writeStream.trig)

5. A helper that catches a bad plan early¶

What we want: if the train might read a box twice (a crash + start-again), and your plan just piles boxes up (append), you'll get double boxes. That's a sneaky bug.

LakeLogic has a little helper robot 🤖 that reads your plan and warns you before it runs:

• STREAM-001: "This plan will make double boxes — use merge instead."
• STREAM-002: "This plan keeps the engine running all day — is that really needed? It costs more."

In [ ]:

from lakelogic.core.contract_lint import review_contract_dict

risky = {
    "info": {"target_layer": "bronze"},
    "source": {"type": "kafka"},
    "trigger": "continuous",
    "materialization": {"strategy": "append"},  # <-- will duplicate on replay
    "model": {"fields": [{"name": "trip_id", "type": "integer"}]},
}
for f in review_contract_dict(risky, "bronze_trip_events"):
    print(f"[{f.severity.upper():8}] {f.check_id}: {f.message}")
    print(f"           fix -> {f.suggestion}")

print()
safe = dict(risky, trigger="available_now", materialization={"strategy": "merge"}, primary_key=["trip_id"])
findings = review_contract_dict(safe, "bronze_trip_events")
print("after fixing (merge + available_now):", findings or "no streaming findings — safe to ship")

from lakelogic.core.contract_lint import review_contract_dict risky = { "info": {"target_layer": "bronze"}, "source": {"type": "kafka"}, "trigger": "continuous", "materialization": {"strategy": "append"}, # <-- will duplicate on replay "model": {"fields": [{"name": "trip_id", "type": "integer"}]}, } for f in review_contract_dict(risky, "bronze_trip_events"): print(f"[{f.severity.upper():8}] {f.check_id}: {f.message}") print(f" fix -> {f.suggestion}") print() safe = dict(risky, trigger="available_now", materialization={"strategy": "merge"}, primary_key=["trip_id"]) findings = review_contract_dict(safe, "bronze_trip_events") print("after fixing (merge + available_now):", findings or "no streaming findings — safe to ship")

6. The big payoff — never count anything twice¶

This is the whole point. Sometimes a crash makes the train read the same boxes again — that's okay, because reading twice is much safer than losing them. The trick is that merge (keyed on trip_id) puts the same ride in the same spot, so reading it again just replaces it. No doubles.

Let's prove it: load some rides, then on purpose read a big overlapping bunch again, and check there are zero duplicate rides at the end.

In [ ]:

import shutil
import os
from deltalake import DeltaTable

target = "lake/bronze_trips_eo"
if os.path.exists(target):
    shutil.rmtree(target)
EO = dict(TRIP_CONTRACT, materialization={"strategy": "merge", "format": "delta", "target_path": target})

# Run 1: trips 0..1999
StreamSink(
    EO,
    trip_events(2000, start=0, bad_every=0),
    engine="polars",
    checkpoint=SQLiteCheckpointStore("eo1.sqlite"),
    checkpoint_key="eo",
    batch_size=1000,
    target_path=target,
).run("available_now")
n1 = DeltaTable(target).to_pyarrow_table().num_rows

# Replay: reprocess ALL of 0..1999 again PLUS 500 new (2000..2499).
# Under bare append this would leave 2000 duplicates; merge keys on trip_id.
StreamSink(
    EO,
    trip_events(2500, start=0, bad_every=0),
    engine="polars",
    checkpoint=SQLiteCheckpointStore("eo2.sqlite"),
    checkpoint_key="eo",
    batch_size=1000,
    target_path=target,
).run("available_now")
tbl = DeltaTable(target).to_pyarrow_table()
ids = tbl.column("trip_id").to_pylist()

print(f"rows after run 1        : {n1}")
print(f"rows after full replay  : {tbl.num_rows}  (2500 distinct, NOT 4500)")
print(f"duplicate trip_ids      : {len(ids) - len(set(ids))}  <- effectively-once")

import shutil import os from deltalake import DeltaTable target = "lake/bronze_trips_eo" if os.path.exists(target): shutil.rmtree(target) EO = dict(TRIP_CONTRACT, materialization={"strategy": "merge", "format": "delta", "target_path": target}) # Run 1: trips 0..1999 StreamSink( EO, trip_events(2000, start=0, bad_every=0), engine="polars", checkpoint=SQLiteCheckpointStore("eo1.sqlite"), checkpoint_key="eo", batch_size=1000, target_path=target, ).run("available_now") n1 = DeltaTable(target).to_pyarrow_table().num_rows # Replay: reprocess ALL of 0..1999 again PLUS 500 new (2000..2499). # Under bare append this would leave 2000 duplicates; merge keys on trip_id. StreamSink( EO, trip_events(2500, start=0, bad_every=0), engine="polars", checkpoint=SQLiteCheckpointStore("eo2.sqlite"), checkpoint_key="eo", batch_size=1000, target_path=target, ).run("available_now") tbl = DeltaTable(target).to_pyarrow_table() ids = tbl.column("trip_id").to_pylist() print(f"rows after run 1 : {n1}") print(f"rows after full replay : {tbl.num_rows} (2500 distinct, NOT 4500)") print(f"duplicate trip_ids : {len(ids) - len(set(ids))} <- effectively-once")

What you just saw¶

• One rule book checks every box — live rides, live prices, and the old pile all use the same rules.
• Crash-safe: the bookmark is saved after the boxes are safely written, so a crash always starts again from the right spot.
• Works on any engine — small ones here, or the big Spark engine — same rules, same safety.
• A helper robot catches double-box plans before they run.
• Never counts twice: reading again is safe, because merge keeps one ride in one spot.

The one thing to remember¶

Every taxi ride gets counted exactly once — even when things crash.

From pretend to real¶

The only pretend parts were the data feeders. Swap them for real ones and everything else stays the same:

In this notebook (pretend)	In real life
KafkaOffsetSource(consumer=InMemoryBroker(...))	KafkaOffsetSource('trips', brokers='...:9092', group_id='rideflow-bronze')
SSEOffsetSource(connect=surge_feed(...))	SSEOffsetSource('https://rideflow/surge/stream')
WatermarkChunkSource(fetch_chunk, ...)	fetch_chunk → a query against your real database
SQLiteCheckpointStore('checkpoints.sqlite')	same file locally; a shared one for the cloud
SparkStreamSink(..., FakeStreamDF)	spark.readStream... on any PySpark platform

Want the full details? See docs/specs/streaming-contracts.md.