Store and retain MQTT time-series data in SQLite

Ingest a small MQTT stream into SQLite and preserve schema, query, late-data, retention, and restore evidence without Docker or a cloud account.

Data Dora: I want you to prove what arrived, what was stored, what a rollup hides, and what must be restored before deletion., your practice guide

Data Dora: I want you to prove what arrived, what was stored, what a rollup hides, and what must be restored before deletion.
Predict the reading, then compare it with the measurement.

Python 3 in your browser (JupyterLite)

Python · no install

Tier 2 · Web · paste-in setup · No account

Version tested: Python 3.12.7 / Pyodide 0.27.6 in JupyterLite 0.6.4; Chromium 148.0.7778.96; captured against a local Mosquitto-compatible WebSocket broker because the capture host cannot reach the public broker; the notebook targets test.mosquitto.org; captureSource playwright:jupyterlite. Date: 2026-09-09.

Open the notebook in your browser and run each Python cell; no install or account is needed.

Three ways to run: use JupyterLite here with no install; run main.py locally from the downloadable lab folder; or open the same notebook in Google Colab.

Open in your browser (new tab)

Steps

Screens captured against JupyterLite Python 3.12.7 / Pyodide 0.27.6; Chromium 148.0.7778.96 on 2026-09-10; the tool may have moved on — the text steps are the contract.

  1. Step 1

    Do
    In the JupyterLite notebook, run the Step 1 cell to create the in-memory readings table and inspect its index contract.
    You will see
    Input: Run python3 main.py --step 1 to create the in-memory readings table and inspect its index contract. Observed heading: "SQLite schema" Observed intermediate: "PRAGMA index_info returned:" Observed rule or check: "1 observed_at" Observed result: "STEP 1 sqlite-schema table=readings index=sensor,observed_at primary-key=event_id" Expected from the existing capture: The terminal prints the CREATE TABLE and CREATE INDEX SQL, then the PRAGMA rows 0 sensor and 1 observed_at.
    Why it matters
    Seeing both schema statements and SQLite's returned index metadata makes identity and observed-time access reviewable before data arrives. Intermediate evidence: "PRAGMA index_info returned:" gives a value or state to inspect. Comparison cue: "1 observed_at" supplies the rule or adjacent evidence for that check. Result: "STEP 1 sqlite-schema table=readings index=sensor,observed_at primary-key=event_id" records how this input was classified.

  2. Step 2

    Do
    In the JupyterLite notebook, run the Step 2 cell to publish six fictional readings to two sensor topics on the public broker.
    You will see
    Six topic payload lines show r1 through r6 arriving on living-room and kitchen topics before Delivery check: six validated payloads reached the subscriber.
    Why it matters
    The actual received topics and JSON establish which sensor identities and fields reached intake behind the final delivery count.

  3. Step 3

    Do
    In the JupyterLite notebook, run the Step 3 cell to validate a payload after deliberately removing its humidity field.
    You will see
    Input: Run python3 main.py --step 3 to validate a payload after deliberately removing its humidity field. Observed heading: "Schema rejection test" Observed intermediate: "{"event_id":"r1","sensor":"living-room","observed_at":100,"received_at":101,"temperature":20.0,"quality":"valid","schema_version":1}" Observed rule or check: "Broken rule: required field check -> missing=humidity" Observed result: "STEP 3 schema-check event=r1 decision=REJECT reason=missing=humidity" Expected from the existing capture: The displayed rejected JSON lacks humidity beside Rule: every Reading field is required before storage. and Broken rule: required field check -> missing=humidity.
    Why it matters
    Putting the malformed payload next to the violated rule lets the learner inspect why rejection is deliberate rather than a silent storage gap. Intermediate evidence: "{"event_id":"r1","sensor":"living-room","observed_at":100,"received_at":101,"temperature":20.0,"quality":"valid","schema_version":1}" gives a value or state to inspect. Comparison cue: "Broken rule: required field check -> missing=humidity" supplies the rule or adjacent evidence for that check. Result: "STEP 3 schema-check event=r1 decision=REJECT reason=missing=humidity" records how this input was classified.

  4. Step 4

    Do
    In the JupyterLite notebook, run the Step 4 cell to insert the validated fixture into SQLite.
    You will see
    Input: Run python3 main.py --step 4 to insert the validated fixture into SQLite. Observed heading: "SQLite write verification" Observed intermediate: "r3 living-room 120 22.0" Observed rule or check: "Duplicate rule: event_id is the primary key." Observed result: "STEP 4 sqlite-write rows=6 sensors=2 duplicate-policy=event-id-primary-key" Expected from the existing capture: After the printed SELECT statement, a six-row table lists each event, sensor, observed time, and temperature; the duplicate rule names event_id as primary key.
    Why it matters
    The returned rows prove what was stored, while the identity rule explains how a replayed event would be constrained. Intermediate evidence: "r3 living-room 120 22.0" gives a value or state to inspect. Comparison cue: "Duplicate rule: event_id is the primary key." supplies the rule or adjacent evidence for that check. Result: "STEP 4 sqlite-write rows=6 sensors=2 duplicate-policy=event-id-primary-key" records how this input was classified.

  5. Step 5

    Do
    In the JupyterLite notebook, run the Step 5 cell to query only the living-room sensor window.
    You will see
    Input: Run python3 main.py --step 5 to query only the living-room sensor window. Observed heading: "Living-room time-series query" Observed intermediate: "r3 120 22.0" Observed rule or check: "3 21.0 20.0 22.0" Observed result: "STEP 5 query sensor=living-room count=3 avg=21.0 min=20.0 max=22.0" Expected from the existing capture: The detail SELECT returns r1=20.0, r2=21.0, and r3=22.0; the aggregate SELECT then returns 3 21.0 20.0 22.0.
    Why it matters
    Showing both SQL statements, source rows, and aggregate row lets the learner recompute the rollup and see what its average hides. Intermediate evidence: "r3 120 22.0" gives a value or state to inspect. Comparison cue: "3 21.0 20.0 22.0" supplies the rule or adjacent evidence for that check. Result: "STEP 5 query sensor=living-room count=3 avg=21.0 min=20.0 max=22.0" records how this input was classified.

  6. Step 6

    Do
    In the JupyterLite notebook, run the Step 6 cell to add one reading whose observed time belongs inside the earlier window but whose receive time is later.
    You will see
    Input: Run python3 main.py --step 6 to add one reading whose observed time belongs inside the earlier window but whose receive time is later. Observed heading: "Late-arrival rollup refresh" Observed intermediate: "before 3 21.0" Observed rule or check: "Decision: label the changed aggregate CORRECTED." Observed result: "STEP 6 late-refresh before=3@21.0 after=4@21.8 status=CORRECTED" Expected from the existing capture: The late-1 row shows observed_at=105 and received_at=205 beside a before/after table changing 3 at 21.0 to 4 at 21.8.
    Why it matters
    The late row and repeated aggregate expose why the result changes and why consumers must receive the CORRECTED label. Intermediate evidence: "before 3 21.0" gives a value or state to inspect. Comparison cue: "Decision: label the changed aggregate CORRECTED." supplies the rule or adjacent evidence for that check. Result: "STEP 6 late-refresh before=3@21.0 after=4@21.8 status=CORRECTED" records how this input was classified.

  7. Step 7

    Do
    In the JupyterLite notebook, run the Step 7 cell to dry-run deletion of readings older than the cutoff; the transaction is rolled back.
    You will see
    Input: Run python3 main.py --step 7 to dry-run deletion of readings older than the cutoff; the transaction is rolled back. Observed heading: "Retention dry run" Observed intermediate: "r2 living-room 110 21.0" Observed rule or check: "Transaction action: ROLLBACK; no candidate was deleted." Observed result: "STEP 7 retention-dry-run candidates=4 min=20.0 max=24.0 would-remain=2 committed=NO" Expected from the existing capture: The retention SELECT lists r1, r4, r2, and r5 as candidates, followed by the DELETE SQL and Transaction action: ROLLBACK; no candidate was deleted.
    Why it matters
    Showing candidate rows before the destructive statement lets the learner audit cutoff scope and confirm the dry run remained reversible. Intermediate evidence: "r2 living-room 110 21.0" gives a value or state to inspect. Comparison cue: "Transaction action: ROLLBACK; no candidate was deleted." supplies the rule or adjacent evidence for that check. Result: "STEP 7 retention-dry-run candidates=4 min=20.0 max=24.0 would-remain=2 committed=NO" records how this input was classified.

  8. Step 8

    Do
    In the JupyterLite notebook, run the Step 8 cell to copy the database with SQLite's backup API and query the restored copy.
    You will see
    Input: Run python3 main.py --step 8 to copy the database with SQLite's backup API and query the restored copy. Observed heading: "SQLite backup restore test" Observed intermediate: "SQL on restored: SELECT COUNT(*) FROM readings -> 6" Observed rule or check: "Comparison: row counts match, so the restored sample is readable." Observed result: "STEP 8 restore-check source-rows=6 restored-rows=6 match=YES" Expected from the existing capture: The source and restored SELECT statements each return 6, followed by Comparison: row counts match, so the restored sample is readable.
    Why it matters
    Running the same read against both databases demonstrates recoverability instead of treating a completed backup call as sufficient evidence. Intermediate evidence: "SQL on restored: SELECT COUNT(*) FROM readings -> 6" gives a value or state to inspect. Comparison cue: "Comparison: row counts match, so the restored sample is readable." supplies the rule or adjacent evidence for that check. Result: "STEP 8 restore-check source-rows=6 restored-rows=6 match=YES" records how this input was classified.

Chapter checks

These questions refer to the chapter’s examples. Use the return links to review their answers.

  1. Which retention release criterion is strongest before deleting raw telemetry?

    Return to the chapter’s knowledge check

  2. Why does a time-partitioned store use chunk- or partition-drop deletion instead of a large DELETE WHERE time < cutoff job?

    Return to the chapter’s knowledge check

  3. A team wants to shorten raw telemetry retention because storage is growing. What should they do before changing the deletion policy?

    Return to the chapter’s knowledge check

Return to Data Retention and Downsampling · Browse Labs