Review MQTT patterns in event time

Publish an out-of-order equipment trace and accept, reject, or hold CEP results using event time, source keys, watermarks, sessions, and explicit uncertainty.

Data Dora: I want you to defend each match with event time, one source key, visible uncertainty, and a bounded output., your practice guide

Data Dora: I want you to defend each match with event time, one source key, visible uncertainty, and a bounded output.
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 read the event contract before correlating the trace.
    You will see
    Input: Run python3 main.py --step 1 and read the event contract before correlating the trace. Observed heading: "CEP event contract" Observed intermediate: "event_second source event time" Observed rule or check: "Rule: correlate only events with the same key in event-time order." Observed result: "STEP 1 event-contract key=equipment-id time=event-second kinds=vibration,current,stop" Expected from the existing capture: The CEP event contract table shows key equipment ID, event_second source event time, and the three accepted event kinds.
    Why it matters
    The visible contract establishes the correlation key, clock, and vocabulary before the learner interprets any sequence. Intermediate evidence: "event_second source event time" gives a value or state to inspect. Comparison cue: "Rule: correlate only events with the same key in event-time order." supplies the rule or adjacent evidence for that check. Result: "STEP 1 event-contract key=equipment-id time=event-second kinds=vibration,current,stop" records how this input was classified.

  2. Step 2

    Do
    In the JupyterLite notebook, run the Step 2 cell to publish the fictional trace to the random public-broker topic.
    You will see
    Three topic payload lines arrive as e2, e1, e3, followed by Notice: e2 arrived before the earlier event-time record e1.
    Why it matters
    The broker-delivered JSON makes arrival order directly observable and exposes why event-time reordering is required.

  3. Step 3

    Do
    In the JupyterLite notebook, run the Step 3 cell to reorder the same trace by event time and evaluate the three-event pattern.
    You will see
    Input: Run python3 main.py --step 3 to reorder the same trace by event time and evaluate the three-event pattern. Observed heading: "Event-time pattern evaluation" Observed intermediate: "e3 28 STOP_REQUEST" Observed rule or check: "Window: [10, 40]; observed gap is 18 seconds." Observed result: "STEP 3 event-time-order=e1>e2>e3 key=conveyor-7 gap=18s pattern=MATCH" Expected from the existing capture: The event-time table shows e1 at 10, e2 at 20, and e3 at 28, then prints Window: [10, 40]; observed gap is 18 seconds.
    Why it matters
    The ordered rows, explicit 30-second rule, and window let the learner reproduce the MATCH decision from the inputs. Intermediate evidence: "e3 28 STOP_REQUEST" gives a value or state to inspect. Comparison cue: "Window: [10, 40]; observed gap is 18 seconds." supplies the rule or adjacent evidence for that check. Result: "STEP 3 event-time-order=e1>e2>e3 key=conveyor-7 gap=18s pattern=MATCH" records how this input was classified.

  4. Step 4

    Do
    In the JupyterLite notebook, run the Step 4 cell to replace the completion event with an otherwise identical event from conveyor-8.
    You will see
    Input: Run python3 main.py --step 4 to replace the completion event with an otherwise identical event from conveyor-8. Observed heading: "Key-isolation check" Observed intermediate: "Rule: every event in one pattern must share the equipment key." Observed rule or check: "Decision: reject e3; keep the conveyor-7 partial state separate." Observed result: "STEP 4 key-check partial=e1>e2 completion=e3@conveyor-8 pattern=REJECT reason=KEY_MISMATCH" Expected from the existing capture: The trace places the e1/e2 partial state on conveyor-7 beside completion: e3@conveyor-8 and says Decision: reject e3.
    Why it matters
    Showing both keys beside the same-key rule explains why completion is rejected while the original partial state remains isolated. Intermediate evidence: "Rule: every event in one pattern must share the equipment key." gives a value or state to inspect. Comparison cue: "Decision: reject e3; keep the conveyor-7 partial state separate." supplies the rule or adjacent evidence for that check. Result: "STEP 4 key-check partial=e1>e2 completion=e3@conveyor-8 pattern=REJECT reason=KEY_MISMATCH" records how this input was classified.

  5. Step 5

    Do
    In the JupyterLite notebook, run the Step 5 cell to compare event e1 at 10 seconds with a 25-second watermark and 20 seconds of allowed lateness.
    You will see
    Input: Run python3 main.py --step 5 to compare event e1 at 10 seconds with a 25-second watermark and 20 seconds of allowed lateness. Observed heading: "Watermark and late-event decision" Observed intermediate: "lateness 15 seconds" Observed rule or check: "Rule: reorder when lateness is within the allowed bound." Observed result: "STEP 5 watermark=25s late-event=e1@10s allowed=20s decision=REORDER corrected-order=e1>e2" Expected from the existing capture: The decision table shows watermark 25, e1 at 10, lateness 15, and allowed lateness 20 before the final decision=REORDER.
    Why it matters
    Laying out the subtraction and bound makes the latency-versus-completeness policy checkable rather than implicit. Intermediate evidence: "lateness 15 seconds" gives a value or state to inspect. Comparison cue: "Rule: reorder when lateness is within the allowed bound." supplies the rule or adjacent evidence for that check. Result: "STEP 5 watermark=25s late-event=e1@10s allowed=20s decision=REORDER corrected-order=e1>e2" records how this input was classified.

  6. Step 6

    Do
    In the JupyterLite notebook, run the Step 6 cell to group event times 0, 5, 12, and 40 seconds with a 15-second inactivity gap.
    You will see
    Input: Run python3 main.py --step 6 to group event times 0, 5, 12, and 40 seconds with a 15-second inactivity gap. Observed heading: "Session windows" Observed intermediate: "session 2: 40" Observed rule or check: "First session closes at 12 + 15 = 27 seconds." Observed result: "STEP 6 session-gap=15s groups=2 sizes=3,1 first-close=27s" Expected from the existing capture: The output lists session 1: 0, 5, 12 and session 2: 40, then computes First session closes at 12 + 15 = 27 seconds.
    Why it matters
    The grouped members and closure arithmetic expose how the 15-second inactivity rule creates two sessions. Intermediate evidence: "session 2: 40" gives a value or state to inspect. Comparison cue: "First session closes at 12 + 15 = 27 seconds." supplies the rule or adjacent evidence for that check. Result: "STEP 6 session-gap=15s groups=2 sizes=3,1 first-close=27s" records how this input was classified.

  7. Step 7

    Do
    In the JupyterLite notebook, run the Step 7 cell to compare an anomaly score of 4.2 with the review threshold.
    You will see
    Input: Run python3 main.py --step 7 to compare an anomaly score of 4.2 with the review threshold. Observed heading: "Anomaly review gate" Observed intermediate: "threshold 3.0" Observed rule or check: "Action: inspect related events without stopping the conveyor automatically." Observed result: "STEP 7 anomaly-score=4.2 threshold=3.0 signal=REVIEW cause=UNPROVEN action=inspect-related-events" Expected from the existing capture: The metric table places score 4.2 over threshold 3.0, then says the score requests review but does not prove cause.
    Why it matters
    Showing the value, threshold, interpretation rule, and bounded inspection action prevents anomaly evidence from becoming an unsupported causal claim. Intermediate evidence: "threshold 3.0" gives a value or state to inspect. Comparison cue: "Action: inspect related events without stopping the conveyor automatically." supplies the rule or adjacent evidence for that check. Result: "STEP 7 anomaly-score=4.2 threshold=3.0 signal=REVIEW cause=UNPROVEN action=inspect-related-events" records how this input was classified.

  8. Step 8

    Do
    In the JupyterLite notebook, run the Step 8 cell to evaluate a gate with true vibration and current conditions but an unknown stop condition.
    You will see
    Input: Run python3 main.py --step 8 to evaluate a gate with true vibration and current conditions but an unknown stop condition. Observed heading: "Three-condition output gate" Observed intermediate: "current TRUE" Observed rule or check: "Rule: PASS only when all three conditions are TRUE." Observed result: "STEP 8 multi-condition key=conveyor-7 vibration=TRUE current=TRUE stop=UNKNOWN decision=HOLD" Expected from the existing capture: The condition table shows vibration TRUE, current TRUE, and stop UNKNOWN before Rule: PASS only when all three conditions are TRUE.
    Why it matters
    The three rows make the missing state explicit and justify HOLD without silently converting uncertainty into pass or failure. Intermediate evidence: "current TRUE" gives a value or state to inspect. Comparison cue: "Rule: PASS only when all three conditions are TRUE." supplies the rule or adjacent evidence for that check. Result: "STEP 8 multi-condition key=conveyor-7 vibration=TRUE current=TRUE stop=UNKNOWN decision=HOLD" 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. A sensor event has a valid source identity but an unknown schema version. What is the strongest first pipeline response?

    Return to the chapter’s knowledge check

  2. After a processor restart, replayed events produce duplicate alerts for the same equipment event. Which part of the pipeline review should be investigated first?

    Return to the chapter’s knowledge check

  3. A dashboard value is published as final, then changes after older event-time records arrive. What is the strongest first review response?

    Return to the chapter’s knowledge check

Return to Building IoT Streaming Pipelines · Browse Labs