{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "> **Read every event once. Survive every crash. Never count anything twice.**" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Real-Time & Streaming — Pipelines That Survive Crashes\n", "\n", "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/LakeLogic/LakeLogic/blob/main/examples/colab/09_streaming_realtime.ipynb) [![View on GitHub](https://img.shields.io/badge/github-view_source-black?logo=github)](https://github.com/lakelogic/LakeLogic/blob/main/examples/colab/09_streaming_realtime.ipynb)\n", "\n", "Think of a little train that carries boxes. Each box is one taxi ride.\n", "\n", "Sometimes the train **crashes**. When it starts again, two bad things used to happen:\n", "\n", "- 😱 It **forgot** some boxes — you lose taxi rides.\n", "- 😱 It **read** some boxes twice — you charge people twice.\n", "\n", "The fix is a well-known trick — it's how big streaming systems (like Spark and Kafka) already stay safe:\n", "\n", "> **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.**\n", "\n", "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.\n", "\n", "The result: **no lost rides, no double rides — every event is handled exactly once.**\n", "\n", "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.\n", "\n", "### The story: RideFlow\n", "\n", "RideFlow is a taxi app. We'll bring its live data in safely, six ways:\n", "\n", "| # | What we want | What makes it happen |\n", "|---|---|---|\n", "| 1 | Read live **taxi rides** and survive a crash — lose nothing, double nothing | `StreamSink` + `KafkaOffsetSource` |\n", "| 2 | Follow a live **price feed** and pick up right where we left off | `SSEOffsetSource` |\n", "| 3 | Load a **giant pile of old rides** without running out of memory | `WatermarkChunkSource` |\n", "| 4 | Do **heavier math on PySpark** (the big engine) | `SparkStreamSink` |\n", "| 5 | **Catch a bad plan** before it makes double rides | streaming checks (`STREAM-001/002`) |\n", "| 6 | Prove we **never double-count**, even after a replay | `merge` + replay |" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Install LakeLogic (Polars + DuckDB engines are all this notebook needs)\n", "!pip install -q lakelogic[polars,duckdb]\n", "\n", "import lakelogic as ll\n", "from lakelogic import (\n", " StreamSink,\n", " SQLiteCheckpointStore,\n", " KafkaOffsetSource,\n", " SSEOffsetSource,\n", " WatermarkChunkSource,\n", " SparkStreamSink,\n", ")\n", "\n", "print(\"LakeLogic\", ll.__version__)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 0. One rule book for every box\n", "\n", "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.\n", "\n", "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)." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "TRIP_CONTRACT = {\n", " \"version\": \"1.0.0\",\n", " \"dataset\": \"bronze_trip_events\",\n", " \"info\": {\"title\": \"bronze_trip_events\"},\n", " \"primary_key\": [\"trip_id\"],\n", " \"model\": {\n", " \"fields\": [\n", " {\"name\": \"trip_id\", \"type\": \"integer\", \"required\": True},\n", " {\"name\": \"rider_id\", \"type\": \"integer\"},\n", " {\"name\": \"fare\", \"type\": \"float\"},\n", " {\"name\": \"surge\", \"type\": \"float\"},\n", " ]\n", " },\n", " \"quality\": {\n", " \"row_rules\": [\n", " {\"name\": \"fare_positive\", \"sql\": \"fare >= 0\"},\n", " ]\n", " },\n", " \"materialization\": {\"strategy\": \"merge\", \"format\": \"delta\", \"target_path\": \"lake/bronze_trips\"},\n", "}\n", "\n", "\n", "def trip_events(n, start=0, bad_every=25):\n", " \"\"\"Simulate n RideFlow trip events; every bad_every-th has a negative (invalid) fare.\"\"\"\n", " out = []\n", " for i in range(start, start + n):\n", " fare = -1.0 if (bad_every and i % bad_every == 0) else round(5 + (i % 40) * 1.5, 2)\n", " out.append({\"trip_id\": i, \"rider_id\": 1000 + (i % 500), \"fare\": fare, \"surge\": 1.0 + (i % 5) * 0.2})\n", " return out\n", "\n", "\n", "print(\"sample event:\", trip_events(1)[0])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. Live taxi rides — crash, then start again with nothing lost\n", "\n", "**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.\n", "\n", "`KafkaOffsetSource` saves the bookmark **after** the boxes are safely written down. If it wakes up later, it jumps straight back to the bookmark.\n", "\n", "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." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# --- A stand-in broker so this notebook runs anywhere (production: pass real brokers) ---\n", "from lakelogic.core.stream_sink import _SimpleTP\n", "\n", "\n", "class _Rec:\n", " __slots__ = (\"topic\", \"partition\", \"offset\", \"value\")\n", "\n", " def __init__(self, t, p, o, v):\n", " self.topic, self.partition, self.offset, self.value = t, p, o, v\n", "\n", "\n", "class InMemoryBroker:\n", " \"\"\"A minimal kafka-python-shaped consumer over in-memory partitions.\"\"\"\n", "\n", " def __init__(self, topic, by_partition):\n", " self.topic = topic\n", " self._data = {p: list(v) for p, v in by_partition.items()}\n", " self._pos = {p: 0 for p in self._data}\n", " self._assigned = []\n", "\n", " def partitions_for_topic(self, t):\n", " return set(self._data)\n", "\n", " def assign(self, tps):\n", " self._assigned = list(tps)\n", "\n", " def seek(self, tp, off):\n", " self._pos[tp.partition] = off\n", "\n", " def poll(self, timeout_ms=None, max_records=500):\n", " out = {}\n", " for tp in self._assigned:\n", " p, vals, start = tp.partition, self._data[tp.partition], self._pos[tp.partition]\n", " if start >= len(vals):\n", " continue\n", " end = min(len(vals), start + max_records)\n", " out[tp] = [_Rec(self.topic, p, o, vals[o]) for o in range(start, end)]\n", " self._pos[p] = end\n", " return out\n", "\n", " def close(self):\n", " pass\n", "\n", "\n", "def kafka_source(by_partition):\n", " broker = InMemoryBroker(\"trips\", by_partition)\n", " return KafkaOffsetSource(\"trips\", consumer=broker, tp_factory=_SimpleTP, max_poll_records=200)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Two partitions of live trips. Drain them through the contract, committing offsets.\n", "ckpt = SQLiteCheckpointStore(\"checkpoints.sqlite\")\n", "\n", "src = kafka_source({0: trip_events(600, start=0), 1: trip_events(400, start=600)})\n", "sink = StreamSink(\n", " TRIP_CONTRACT,\n", " src,\n", " engine=\"polars\",\n", " checkpoint=ckpt,\n", " checkpoint_key=\"trips\",\n", " batch_size=250,\n", " target_path=\"lake/bronze_trips\",\n", ")\n", "summary = sink.run(\"available_now\") # AvailableNow: drain to current end, then exit\n", "\n", "print(f\"batches : {summary.batches}\")\n", "print(f\"events read : {summary.source_count}\")\n", "print(f\"valid (good): {summary.good_count}\")\n", "print(f\"quarantined : {summary.bad_count}\")\n", "print(f\"committed offset (cursor): {summary.cursor}\") # per-partition broker offsets" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "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." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# The topic has grown: 200 more events arrived on partition 0 since we last ran.\n", "src2 = kafka_source({0: trip_events(800, start=0), 1: trip_events(400, start=600)})\n", "sink2 = StreamSink(\n", " TRIP_CONTRACT,\n", " src2,\n", " engine=\"polars\",\n", " checkpoint=ckpt,\n", " checkpoint_key=\"trips\",\n", " batch_size=250,\n", " target_path=\"lake/bronze_trips\",\n", ")\n", "resumed = sink2.run(\"available_now\")\n", "\n", "print(f\"resumed from : {resumed.resumed_from}\") # seeks to the committed offsets\n", "print(f\"events read : {resumed.source_count}\") # ONLY the 200 new ones — no reprocessing\n", "print(f\"new cursor : {resumed.cursor}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. A live price feed — pick up right where we left off\n", "\n", "**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.\n", "\n", "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.\n", "\n", "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." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def surge_feed(events):\n", " \"\"\"Fake SSE server: replays every event AFTER the armed Last-Event-ID.\"\"\"\n", " import json\n", "\n", " class Evt:\n", " __slots__ = (\"id\", \"data\")\n", "\n", " def __init__(self, i, d):\n", " self.id, self.data = i, d\n", "\n", " def connect(last_event_id):\n", " started = last_event_id is None\n", " for eid, payload in events:\n", " if not started:\n", " if str(eid) == str(last_event_id):\n", " started = True\n", " continue\n", " yield Evt(str(eid), json.dumps(payload))\n", "\n", " return connect\n", "\n", "\n", "price_events = [\n", " (i, {\"trip_id\": i, \"rider_id\": 1000 + i, \"fare\": 12.0, \"surge\": 1.0 + (i % 6) * 0.3}) for i in range(300)\n", "]\n", "\n", "sse_ckpt = SQLiteCheckpointStore(\"checkpoints.sqlite\")\n", "sse = SSEOffsetSource(connect=surge_feed(price_events))\n", "run1 = StreamSink(\n", " TRIP_CONTRACT,\n", " sse,\n", " engine=\"polars\",\n", " checkpoint=sse_ckpt,\n", " checkpoint_key=\"surge\",\n", " batch_size=100,\n", " target_path=\"lake/bronze_trips\",\n", ").run(\"available_now\")\n", "print(\"drained events:\", run1.source_count, \"| Last-Event-ID committed:\", run1.cursor)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Feed grows to 480 ticks; reconnect resumes from the committed Last-Event-ID.\n", "price_events_grown = [\n", " (i, {\"trip_id\": i, \"rider_id\": 1000 + i, \"fare\": 12.0, \"surge\": 1.0 + (i % 6) * 0.3}) for i in range(480)\n", "]\n", "sse2 = SSEOffsetSource(connect=surge_feed(price_events_grown))\n", "run2 = StreamSink(\n", " TRIP_CONTRACT,\n", " sse2,\n", " engine=\"polars\",\n", " checkpoint=sse_ckpt,\n", " checkpoint_key=\"surge\",\n", " batch_size=100,\n", " target_path=\"lake/bronze_trips\",\n", ").run(\"available_now\")\n", "print(\"resumed from Last-Event-ID:\", run2.resumed_from)\n", "print(\"only new ticks read :\", run2.source_count, \"(events 300..479)\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. Load a giant pile of old rides — without running out of memory\n", "\n", "**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.\n", "\n", "`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.\n", "\n", "You tell it how to grab one handful (`fetch_chunk`), and that can be any database or file — LakeLogic doesn't care which." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Simulate a 5,000-row historical table behind a keyset query.\n", "history = trip_events(5000, start=0, bad_every=0) # all valid, for a clean count\n", "history_sorted = sorted(history, key=lambda r: r[\"trip_id\"])\n", "\n", "\n", "def fetch_chunk(after, limit):\n", " if after is None:\n", " start = 0\n", " else:\n", " start = next((i for i, r in enumerate(history_sorted) if r[\"trip_id\"] > after), len(history_sorted))\n", " return history_sorted[start : start + limit]\n", "\n", "\n", "backfill_ckpt = SQLiteCheckpointStore(\"checkpoints.sqlite\")\n", "backfill = WatermarkChunkSource(fetch_chunk, watermark_field=\"trip_id\", chunk_size=500)\n", "b = StreamSink(\n", " TRIP_CONTRACT,\n", " backfill,\n", " engine=\"polars\",\n", " checkpoint=backfill_ckpt,\n", " checkpoint_key=\"backfill\",\n", " batch_size=500,\n", " target_path=\"lake/bronze_trips\",\n", ").run(\"available_now\")\n", "print(f\"rows loaded : {b.source_count} (in {b.batches} bounded chunks of 500 — flat memory)\")\n", "print(f\"watermark : {b.cursor} (last trip_id; resume continues strictly after this)\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. Bigger math on the big engine (PySpark)\n", "\n", "**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.\n", "\n", "On any PySpark platform (Databricks, EMR, Dataproc, Synapse, or your own Spark cluster) it looks like this:\n", "\n", "```python\n", "silver_events = (spark.readStream.format('delta').load('lake/bronze_trips'))\n", "\n", "sink = SparkStreamSink(\n", " SILVER_SURGE_CONTRACT,\n", " silver_events,\n", " checkpoint_location='/checkpoints/silver_surge', # Spark keeps the bookmark\n", " trigger='available_now', # or run all the time: trigger='processing_time', processing_time='30 seconds'\n", " on_batch=lambda b: print(f'batch {b.batch_id}: good={b.good_count} bad={b.bad_count}'),\n", ")\n", "query = sink.run()\n", "```\n", "\n", "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:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Prove the foreachBatch handler runs the contract per micro-batch (no Spark needed).\n", "# The ONLY fake here is the writeStream wiring; the micro-batches are real frames\n", "# and each one flows through the real contract engine. (We use engine='polars' so\n", "# it runs locally; in production SparkStreamSink defaults to engine='spark'.)\n", "import polars as pl\n", "\n", "\n", "class FakeWriter:\n", " def __init__(self, batches):\n", " self._b, self.opts, self.trig = batches, {}, None\n", "\n", " def foreachBatch(self, fn):\n", " self._fn = fn\n", " return self\n", "\n", " def option(self, k, v):\n", " self.opts[k] = v\n", " return self\n", "\n", " def outputMode(self, m):\n", " return self\n", "\n", " def trigger(self, **kw):\n", " self.trig = kw\n", " return self\n", "\n", " def start(self):\n", " for i, frame in enumerate(self._b):\n", " self._fn(frame, i) # Spark drives foreachBatch\n", " return self\n", "\n", " def awaitTermination(self):\n", " pass\n", "\n", "\n", "class FakeStreamDF:\n", " def __init__(self, batches):\n", " self.writeStream = FakeWriter(batches)\n", "\n", "\n", "micro_batches = [pl.DataFrame(trip_events(200, start=0)), pl.DataFrame(trip_events(200, start=200))]\n", "df = FakeStreamDF(micro_batches)\n", "spark_sink = SparkStreamSink(\n", " TRIP_CONTRACT,\n", " df,\n", " engine=\"polars\",\n", " checkpoint_location=\"/checkpoints/silver_surge\",\n", " target_path=\"lake/silver_surge\",\n", " on_batch=lambda b: print(f\" micro-batch {b.batch_id}: good={b.good_count} bad={b.bad_count}\"),\n", ")\n", "spark_sink.run()\n", "print(\"checkpointLocation (Spark owns resume):\", df.writeStream.opts[\"checkpointLocation\"])\n", "print(\"trigger:\", df.writeStream.trig)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. A helper that catches a bad plan early\n", "\n", "**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.\n", "\n", "LakeLogic has a little helper robot 🤖 that reads your plan and warns you *before* it runs:\n", "\n", "- **`STREAM-001`**: \"This plan will make double boxes — use `merge` instead.\"\n", "- **`STREAM-002`**: \"This plan keeps the engine running all day — is that really needed? It costs more.\"" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from lakelogic.core.contract_lint import review_contract_dict\n", "\n", "risky = {\n", " \"info\": {\"target_layer\": \"bronze\"},\n", " \"source\": {\"type\": \"kafka\"},\n", " \"trigger\": \"continuous\",\n", " \"materialization\": {\"strategy\": \"append\"}, # <-- will duplicate on replay\n", " \"model\": {\"fields\": [{\"name\": \"trip_id\", \"type\": \"integer\"}]},\n", "}\n", "for f in review_contract_dict(risky, \"bronze_trip_events\"):\n", " print(f\"[{f.severity.upper():8}] {f.check_id}: {f.message}\")\n", " print(f\" fix -> {f.suggestion}\")\n", "\n", "print()\n", "safe = dict(risky, trigger=\"available_now\", materialization={\"strategy\": \"merge\"}, primary_key=[\"trip_id\"])\n", "findings = review_contract_dict(safe, \"bronze_trip_events\")\n", "print(\"after fixing (merge + available_now):\", findings or \"no streaming findings — safe to ship\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 6. The big payoff — never count anything twice\n", "\n", "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.\n", "\n", "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." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import shutil\n", "import os\n", "from deltalake import DeltaTable\n", "\n", "target = \"lake/bronze_trips_eo\"\n", "if os.path.exists(target):\n", " shutil.rmtree(target)\n", "EO = dict(TRIP_CONTRACT, materialization={\"strategy\": \"merge\", \"format\": \"delta\", \"target_path\": target})\n", "\n", "# Run 1: trips 0..1999\n", "StreamSink(\n", " EO,\n", " trip_events(2000, start=0, bad_every=0),\n", " engine=\"polars\",\n", " checkpoint=SQLiteCheckpointStore(\"eo1.sqlite\"),\n", " checkpoint_key=\"eo\",\n", " batch_size=1000,\n", " target_path=target,\n", ").run(\"available_now\")\n", "n1 = DeltaTable(target).to_pyarrow_table().num_rows\n", "\n", "# Replay: reprocess ALL of 0..1999 again PLUS 500 new (2000..2499).\n", "# Under bare append this would leave 2000 duplicates; merge keys on trip_id.\n", "StreamSink(\n", " EO,\n", " trip_events(2500, start=0, bad_every=0),\n", " engine=\"polars\",\n", " checkpoint=SQLiteCheckpointStore(\"eo2.sqlite\"),\n", " checkpoint_key=\"eo\",\n", " batch_size=1000,\n", " target_path=target,\n", ").run(\"available_now\")\n", "tbl = DeltaTable(target).to_pyarrow_table()\n", "ids = tbl.column(\"trip_id\").to_pylist()\n", "\n", "print(f\"rows after run 1 : {n1}\")\n", "print(f\"rows after full replay : {tbl.num_rows} (2500 distinct, NOT 4500)\")\n", "print(f\"duplicate trip_ids : {len(ids) - len(set(ids))} <- effectively-once\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## What you just saw\n", "\n", "- **One rule book** checks every box — live rides, live prices, and the old pile all use the same rules.\n", "- **Crash-safe**: the bookmark is saved *after* the boxes are safely written, so a crash always starts again from the right spot.\n", "- **Works on any engine** — small ones here, or the big Spark engine — same rules, same safety.\n", "- **A helper robot** catches double-box plans before they run.\n", "- **Never counts twice**: reading again is safe, because `merge` keeps one ride in one spot.\n", "\n", "### The one thing to remember\n", "\n", "> **Every taxi ride gets counted exactly once — even when things crash.**\n", "\n", "### From pretend to real\n", "\n", "The only pretend parts were the data feeders. Swap them for real ones and everything else stays the same:\n", "\n", "| In this notebook (pretend) | In real life |\n", "|---|---|\n", "| `KafkaOffsetSource(consumer=InMemoryBroker(...))` | `KafkaOffsetSource('trips', brokers='...:9092', group_id='rideflow-bronze')` |\n", "| `SSEOffsetSource(connect=surge_feed(...))` | `SSEOffsetSource('https://rideflow/surge/stream')` |\n", "| `WatermarkChunkSource(fetch_chunk, ...)` | `fetch_chunk` → a query against your real database |\n", "| `SQLiteCheckpointStore('checkpoints.sqlite')` | same file locally; a shared one for the cloud |\n", "| `SparkStreamSink(..., FakeStreamDF)` | `spark.readStream...` on any PySpark platform |\n", "\n", "Want the full details? See [`docs/specs/streaming-contracts.md`](https://github.com/lakelogic/LakeLogic/blob/main/docs/specs/streaming-contracts.md)." ] } ], "metadata": { "colab": { "provenance": [] }, "kernelspec": { "display_name": "Python 3", "name": "python3" }, "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 0 }