DEV Community: Jennifer Evangelista

Night Guard: an open-model guard that watches out for kids in game chats, so tired parents don't have to read every message

Jennifer Evangelista — Sat, 24 Oct 2026 03:05:14 +0000

This is a submission for the Hacktoberfest 2026 Week 3 Challenge: Slow and Steady.

Looking for the prize categories? One table at the end (Prize Categories) lists each one, what I built for it and which section shows it.

What I Built

Parents can't read every chat their kids have in Roblox, Discord or TikTok. Meanwhile the US National Center for Missing & Exploited Children received more than 50,000 reports of financial sextortion in 2025, about 137 a day (NCMEC, 2026). And it can be fast: in a Swansea University study of chats between groomers and researchers posing as children, the time from first contact to the first sexual request ranged from 18 minutes to 82 hours (Irish Examiner, 2016).

Night Guard is a small open-model guard designed to reduce that review effort. It reads messages on the family's own computer and highlights possible pressure tactics: someone asking a kid to keep a secret, to move to a private app, to send a photo, or threatening to share one.

It talks to two different people in two different ways:

To the kid, it's a friend that turns them into a hero. It warns in plain words and says "danger" when it is one: "DANGER: someone asked you to hide this from your family. It's like a stranger on the street saying 'come with me, but don't tell your mom'. You don't go, and you tell an adult." A calm voice can read it aloud. The kid earns practice progress from completed exercises, while previously practiced explanations can be reopened. Repeated exposure to warnings does not certify learning, and protection stays on.
To the parents, it's the Night Guard 🌙. They get the risk level, the app and the flagged phrase (never the whole chat), why it matters, and what to do right now. Optionally, a push on their phone.

The intended lazy payoff is less repetitive review for parents and shorter, useful explanations for kids. The guard highlights events for the family and repeats fewer lessons as the child practices. Time saved and real-world protection have not been measured.

What it is not

It's not a bad-words filter and not sentiment analysis. Pressure can appear in friendly wording ("you're so mature for your age", "free Robux for you"). Night Guard uses an open model to read the messages provided to each check and match it against 14 app-defined tactic labels. Examples of these behaviors are documented in Public Safety Canada's 2025 guidance on gaming platforms, NCMEC's sextortion guidance, and Thorn's Online Grooming Report (2022). The live overlay supplies the 12 most recent messages, not the full chat history.

Demo

The short film (2:30), narrated by Snoozy, the lazy night watchman:

👉 Try it: https://night-guard-demo.onrender.com (fictional examples only; the free instance can take ~50 s to wake up, a check takes about 7 to 15 s online, and it accepts a limited number of checks per hour so it stays free for everyone)

Start with ① Watch the guard act on its own: a fictional game chat where a stranger writes and Night Guard warns the kid by itself, without anyone pasting anything.

Then ③ See what parents get:

Every feature, recorded live on the local app (about 3 minutes):

The online demo runs the Tinker fine-tuned model in the cloud so judges can try it without installing anything. The real Night Guard runs on the family's computer with Gemma, where chat analysis stays on the device. Optional external link inspection is a separate action described below.

Code

jenniferevangelista-de / night-guard

Helps kids understand what happened in an online chat and choose how to ask for help. Runs locally with Gemma.

Video production notes and credits.

The current full suite has 94 passing tests (pytest tests), including the demo's request limits, regression cases for every bug found in testing and simulated tests for the link inspector and Backboard client. Separate live Auth0, Browserbase and SerpApi checks are described below; automated tests do not replace those integration checks. The offline test run and validation of all 30 audio assets are recorded in the repository.

How I Built It

Auth0 sign-in for the local Family controls is optional and only applies when configured; the public demo does not use Auth0.

The detector: Gemma 4, on the kid's computer

The core is gemma4:e4b running locally through Ollama on a laptop GPU (RTX 4050, 6 GB). Each check sends the recent messages with a structured JSON schema and gets back the signals it found, the message index and the exact quote.

I don't trust the model blindly. Every answer goes through checks before a kid or parent sees it:

An unreadable answer is never "all clear". If the model is down or answers in a broken format, the kid sees "Couldn't analyze this chat. This is NOT an all-clear", never a green light. The evaluations also count a failed analysis as an error, never as a catch or as a correct "safe".
Evidence is checked against the chat. The quote must exist in a message from the contact; a paraphrased quote is replaced by the real message; the kid's own messages are never flagged.
Two narrow tactics need their own words in the chat. "Came back with a new account" only counts if the chat mentions being blocked or ignored, and "you're so mature" flattery only counts with maturity or age words. Before this, a classmate who lost his password and a compliment on a Minecraft build were flagged.

Results on a fictional test set of 70 chats (36 risky, 34 safe), written by me for testing:

	First version (60 chats, 11 signals)	Now (70 chats, 14 signals)
Valid analyses (an error never counts as a catch)	not measured	70/70
Risky chats flagged	30/30	36/36
False alarms on safe chats	4/30 (13%)	3/34 (9%)
Signal precision / recall	89% / 87%	88% / 87%
Avg time per check	3.8 s (historical run)	3.41 s (local run; the result file is not versioned)

The three remaining false alarms are on purpose hard cases: an in-game "secret" about a leaked Halloween event, a classmate saying "see you at school", and "how old r u now" on a birthday. Honesty note: two newer hard cases were fixed after I saw them fail, so they're no longer a blind test. The 60-chat and 70-chat columns are different development evaluations, and the 70 cases have already informed changes. The 36/36 result applies to these fictional patterns, not an independent holdout or real-world protection.

Proactive: it doesn't wait for a kid to paste anything

A "paste your chat here" tool only helps if the kid thinks to use it, and pressure can escalate within a single chat session. So Night Guard also watches as messages arrive:

Screen guard with Gemma's vision. Kids chat inside installed apps (the Roblox app, the Discord app, Minecraft). Every few seconds Night Guard checks which program owns the window in front (by its process, not its title, so a browser tab or a document called "Roblox" is ignored). Only if it is one of those apps does it take a screenshot in memory, Gemma 4 reads the chat from the image locally (~7 s warm), and the text goes to the same detector. The image is never saved and never leaves the computer. Banking, email or school windows are not on the list, so they are never captured. If the window changes before the capture, nothing is read.
A Chrome extension for web chats (Discord web and WhatsApp Web, experimental), which only talks to the local server and refuses any other address.
A "🛡️ Guard on" badge identifies the active guard, and it turns into "Guard paused: can't analyze right now" when analysis fails; silence is not proof of safety. Kids know it's there. In a study of 736 reviews attributed to children aged 8–19, 76% gave safety apps one star (Ghosh et al.). That finding motivates transparent design; it does not establish that covert monitoring causes harm to family relationships.

Fine-tuning with Tinker: teaching a model to speak the app's language

Tinker doesn't serve Gemma, so for the cloud demo I fine-tuned Qwen3-8B with a LoRA (rank 16) on Thinking Machines' Tinker, using fictional training chats that share no sentence with the test set. Two runs: v1 (11 tactics, 800 chats) and v2 (14 tactics, 1,000 chats, ~1.23M tokens), all compared on the same 70 test chats:

	Qwen3-8B base	LoRA v1	LoRA v2 (online demo)
Answers that follow the app's full schema (tactic id, message index, quote, summary)	11 / 70	70 / 70	70 / 70
Risky chats flagged (unusable answers never count as catches)	0 / 36 (none of the 36 was usable)	36 / 36	36 / 36
False alarms on safe chats	0 / 11 valid (23 invalid)	3 / 34	2 / 34
Signal precision / recall	n/a (no valid risky answer)	0.84 / 0.79	0.88 / 0.83
Avg time per check	3.0 s	5.0 s	4.5 s

Latency values are the aggregate figures stored in the current result JSON files. Older run logs report different timings; per-case timings were not retained, so these averages cannot be independently recomputed.

Strict format conformity (valid_schema) is reported separately from usable coverage. The common postprocessor retains valid evidence in a partial answer; an unusable answer is an error, never a catch or a correct safe result. These saved Tinker runs contain no partial answers, so strict-schema and usable-coverage totals coincide: 11/70 for base and 70/70 for each LoRA. The base model often renames fields (signal_id instead of id). Read leniently, it flagged 35 of the 35 risky chats it answered readably, so it often sees the pattern; its unrecognized fields made those answers unusable under the original schema. Fine-tuning produced schema-valid answers on all 70 development chats and found more signals, including the three newer tactics in v2. The online demo runs v2. Two honest limits: the false-alarm gap between v1 and v2 (3 vs 2 of 34) is a single chat, and the LoRA is not faster than the base model; this is a measured format improvement on the evaluated cases, not a guarantee for future inputs. Every raw answer is saved per case in the repo, so anyone can recount.

Temporal: an evaluation that survives a crashing laptop

Evaluating a local model on a laptop is fragile. The evaluation runs as a Temporal workflow where each conversation is an activity with automatic retries. In a 60-conversation crash test, Ollama was stopped during the run. The saved eval/temporal_report.json records s12 completing on attempt 5. The failed activity was retried when the model returned; completed activities remained in the workflow history.

ElevenLabs: a calm voice for kids who don't read fast yet

Younger kids and kids with dyslexia may not read a warning in time. Every child warning (14 tactics × English and Portuguese) was voiced once with ElevenLabs (Sarah voice, eleven_multilingual_v2) and shipped as audio files. In the live game demo and in the screen guard popup, the guard reads the most serious new warning aloud, and it speaks again if the stranger insists with a new message. Only the fixed warning texts were ever sent to ElevenLabs. No chat, ever.

Sentry: tracing the agent without reading the chats

When configured, Sentry traces each check as an agent run (night-guard.check) with a model call inside (gen_ai.chat) and the evidence step after it. That's how I saw that a cold Gemma call takes up to 8 s and a warm one about 4.5 s on this laptop. The traces carry only numbers and labels: model, backend, latency, number of messages, severity and tactic ids. Never a message, a name or a quote: send_default_pii is off, local variables are not captured, and every event goes through an allowlist: only fields known to hold numbers and labels are kept, everything else (request bodies, breadcrumbs, exception text, any new field) is dropped. A test pushes a fake event full of a marker string and checks the marker never survives. The UI's status polling is sampled out so the agent runs stand out.

Render: one click for the judges

The public demo runs on Render (Blueprint render.yaml, free instance), with the Tinker fine-tuned model as its AI backend. The app does not persist chats on the demo server: family rules and alerts stay in each visitor's own browser, and the screen says to use fictional examples only. This describes application behavior, not the retention policies of Render or the model provider.

Browserbase + SerpApi: check a link without opening it on the child's computer

When a conversation contains a link, "Check without opening" uses Browserbase to open it in an isolated cloud
browser. The inspector reads the final destination, checks for password fields and returns a screenshot.
SerpApi separately searches for reports about the domain. The selected URL goes to Browserbase and the domain
to SerpApi; the surrounding conversation is never sent. A URL can itself contain personal information.
Each provider is optional, and the result identifies the checks that ran. Missing reports never mean "safe".

Live local evidence, October 23, 2026: one Browserbase session opened the fictional BlockQuest "free gems"
page (session creation HTTP 201, page HTTP 200). It detected the password field, captured a screenshot and
returned the Portuguese warning "Não abra esse link" ("Don't open this link") in 12.68 seconds. Session release
returned HTTP 200. A separate, single SerpApi query for night-guard-demo.onrender.com returned HTTP 200 and six
search results, with none matching the report heuristic. The fictional page submits no credentials.

The deployed Render endpoint also passed a live check after the viewport correction: POST /api/link/check
returned HTTP 200, identified both Browserbase and SerpApi as checked, detected the password field and returned
the same warning in 8.90 s. Its domain search returned six results, with none matching the report heuristic.
The returned screenshot was visually verified to show the complete fictional login form; that public-endpoint
screenshot is shown above.

These checks establish provider connectivity and the observed password detection on one fictional example,
not scam-detection accuracy.

Optional Auth0 sign-in for local Family controls

When configured, the local Ollama app uses Auth0 to sign parents in before they can read family alerts, change rules or stop
the guard. The email must be verified and present in the family's PARENT_EMAILS list. It requires Auth0
application settings and a session secret; without configuration, the local Family view has no login gate.
The public demo does not use Auth0.

Live local evidence: on a separate test server at port 8766, the alerts endpoint returned HTTP 401 before
login, and the login route redirected with HTTP 302. A real Google sign-in with a verified, allowlisted parent
completed the callback and the Family view displayed the signed-in parent. Logout returned home and restored
the locked Parents view and sign-in button. Rejection of unverified or unlisted accounts is unit-tested; a real
rejected-account attempt has not been performed. Local servers need restarting after .env changes.

Backboard: compare open-weight models through one API key

On October 23, 2026, Backboard served three complete evaluations: Gemma 3 27B, Llama 3.3 70B and Mistral
Small 3.1 24B, each on the same 70 fictional development conversations (36 risky, 34 safe). All three received
the shared classification SYSTEM and the canonical app SCHEMA explicitly in the user prompt, with JSON mode,
memory off, a fresh conversation per case and unchanged app postprocessing/scoring. An initial smoke returned
{} when only JSON mode was requested; adding the same format schema for every model fixed that missing
instruction. No gold labels were added to prompts. This is a common Backboard adapter, not the identical
prompt/decoding setup used by the older Tinker baseline or schema-constrained Ollama.

Model	Strict schema	Usable coverage	Risky chats flagged	False alarms on valid safe cases	Signal precision	Signal recall
Gemma 3 27B	68/70	69/70	36/36	3/33 (1 error)	0.938	0.849
Llama 3.3 70B	69/70	69/70	36/36	3/33 (1 error)	0.912	0.981
Mistral Small 3.1 24B	70/70	70/70	36/36	3/34	0.852	0.981
Qwen 2.5 72B	n/a	n/a	n/a	n/a	Unavailable in live smoke	No full result

Strict-schema validity and usable coverage are separate. The shared scorer accepts a partial analysis when
postprocessing retains valid evidence; an unusable analysis is an error, never a catch or a correct safe answer.
Gemma's r06 contained two valid personal-information items and one malformed empty item: the empty item was
discarded and the partial analysis remained usable. Its safe case s33 was unusable. Llama had one unusable safe
case. These failures remain in the denominators and the saved raw answers; nothing was retried to improve scores.

The practical difference here is format reliability and signal precision/recall, not a demonstrated safety
advantage: all three flagged the 36 risky development cases and raised three false alarms. Mistral followed
the strict schema on every case; Gemma had higher signal precision but lower recall. This is a single run on
reused development cases, not a new blind holdout or proof of protection for real children. Request timings
include shared client slots and waiting, so they are not compared with local Gemma or Tinker latency.

Qwen 2.5 72B's smoke request hit a 120.65-second read timeout. An already-started automatic retry was canceled
locally when the other models were moved to a shared four-request limit. Both Qwen requests appear in billing,
but no answer was observed locally and no 70-case Qwen result is claimed. The model was not silently replaced.

Backboard's usage API recorded USD 0.034062 across 216 events, including 210 scored calls, three successful
smokes, the initial format-debugging call and two Qwen requests. A conservative token-price/reservation budget
tracked USD 0.065078 against the internal USD 1 cap; that estimate is separate from actual billing. Only
fictional conversations were sent. The raw results and summary
preserve per-case outputs, errors, coverage and signal metrics.

What the family gets

Family rules: for each tactic, "Tell family" or "Child decides". Photo requests, meeting in person, password requests, secrecy, coming back after a block and sextortion go to the family by default. The kid is always told when that happens: "An alert was recorded for your family. Tell a trusted adult; this is not a delivery confirmation. You're not in trouble."
The gap no platform covers: Instagram supervision, Discord Family Center, TikTok Family Pairing and Roblox parental controls each protect their own app. The classic path starts in a game and moves to a private app. Night Guard points out a possible match: "Same name seen in more than one app (possibly the same person): Alex_17 (Roblox → Instagram)". It's a hint to check, not proof, because names can repeat.
What to do now, by severity: talk calmly and without punishment (kids who get in trouble stop telling), keep screenshots, block and report, and for image requests, threats or money demands, report to SaferNet Brasil or the NCMEC CyberTipline and the police. Don't pay.
Phone alerts through ntfy (free, no account) with only the risk level, the app and the kind of risk.
"Tell someone": the kid picks what to share and gets a ready message for Mom, Dad or a teacher. It is assembled deterministically from their own words and the flagged quotes.

Limits and next step

Phones are the next milestone. Today Night Guard protects a Windows computer. On Android, the plan is an app that reads message notifications (the permission parental tools use there) and runs the detector on-device. On iPhone, Apple doesn't let any app read other apps' messages, so the realistic path is Apple's own Communication Safety.
It's a learning and support tool, not a guarantee of safety. It can miss things and raise false alarms. It doesn't replace trusted adults or platform reporting.
All test data is fictional. No real children's conversations were used anywhere.

Why Does Open Innovation Matter?

Child safety is the clearest case for open models I know. A guard that reads a kid's private chats should run on the family's own computer, be inspectable, and be free. For comparison, Bark Premium lists US$14 a month before tax. Platforms protect only their own walls. Roblox open-sourced Sentinel in 2025 and announced an update and ROOST contribution in 2026, which is useful for platforms; Night Guard is the other half: an open, local tool that sits with the family and works across apps.

Open weights made every part of this possible: Gemma reading text and screenshots on a laptop GPU, a Qwen LoRA fine-tuned in an afternoon, and anyone able to audit exactly which 14 tactics trigger a warning and what words a kid will see.

The Lazy Payoff

Parents can review flagged events with suggested next steps, and stay the ones who talk and decide. Reduced review effort is an intended benefit, not a measured result.
Kids get short explanations and practice choices, and can ask an adult for help. Progress in the app does not establish readiness for future situations.
The guard can repeat fewer explanations after completed practice, while leaving every explanation available to reopen. Completing an exercise does not prove future safety and never switches protection off.

Prize Categories

Each row: what the category asks for, what I built, and the section that shows it.

Category	What it asks	What I built	Where to see it
Gemma	Use Gemma: run it locally, fine-tune it or serve it	`gemma4:e4b` on the family's computer reads the chat and, through screenshots, the game window	The detector: Gemma 4; Proactive
Tinker	Fine-tune a model and show a clear improvement over a baseline	Qwen3-8B LoRA: 70/70 answers in the app's schema, against 11/70 for the base model	Fine-tuning with Tinker
Render	Use Render as the AI runtime or to host the front end	The public demo runs the fine-tuned model, deployed from a Blueprint	Render: one click for the judges
Temporal	Make an agent durable: survive failures and retry	The evaluation runs as a workflow; stopping the model mid-run lost nothing	Temporal: an evaluation that survives a crashing laptop
ElevenLabs	Give an open-source agent a voice	Spoken child warnings in English and Portuguese, shipped as audio files	ElevenLabs: a calm voice for kids
Sentry Agent Tracing	Show the agent's work with traces or screenshots	Agent runs and model calls traced, with an allowlist so no chat text is sent	Sentry: tracing the agent without reading the chats
Browserbase	Give an agent a real browser	The link inspector opens a link in an isolated cloud browser (Playwright over CDP, not Stagehand) and finds password fields	Browserbase + SerpApi
SerpApi	Give an agent live web search	The same inspector searches reports about the link's domain; no reports is never read as "safe"	Browserbase + SerpApi
Auth0 for AI Agents	Sign users in to an agent	Verified, allowlisted Google login for the Family controls. Local app only, not the public demo	Optional Auth0 sign-in
Backboard	Compare open-weight models through one API key	Three models on the same 70 fictional chats through one key. A Qwen timeout is disclosed	Backboard: compare open-weight models
Overall prize	Writing, theme, creativity, technical execution	A friend for the kid, a night guard for the parents; a failure is shown as "couldn't analyze", never "all clear"	The whole post

These entries describe the implemented uses and their observed evidence. Final category eligibility is for
the judges under the challenge criteria.

Your pipeline ran green ✅ and shipped last week’s data anyway

Jennifer Evangelista — Fri, 23 Oct 2026 10:12:57 +0000

Image generated with AI

Inspired by the “Is this a pigeon?” meme. The dates and scenario are illustrative.

Your backfill finished at 07:03 this morning. It ran cleanly. Every dependency was green. But it was processing last Thursday’s data, and your gate just let it through as today’s load.

A successful backfill can finish this morning and still belong to last week’s reporting period. If the downstream gate checks only whether its dependencies ran today, the wrong load can pass.

I would make business date an explicit input and retain the exact upstream run IDs that authorized the output. The small SQLite lab accompanying this article demonstrates that contract, including a retry after publication and two concurrent publishers.

The clock answers a narrower question

Imagine two independently scheduled dimension jobs and a daily sales load. The sales load needs both dimensions for October 23.

At 07:00 UTC on October 23, both jobs complete successfully. A check based on today’s start time sees two green runs and opens the gate. But those runs were backfills for October 16. The current reporting period is still missing.

The opposite case is just as useful: a valid October 23 run can start before midnight and complete later. A midnight cutoff can exclude work the consumer actually needs.

In the Databricks Jobs API, start_time_from filters runs by when they started, which says nothing about the business period the code processed. The API also separates completed runs from successful ones. Both are good inputs for a collector, but neither tells you a run covered the right day.

The fix is not a smarter clock. It is an explicit record of which business date each run actually processed, and a gate that checks that record instead of the wall clock.

Write down which attempt counts

My example records job ID, business date, attempt ordinal, run ID, state and a timezone-aware start timestamp. It checks the exact required job IDs against the requested date.

The attempt ordinal is assigned by the producer for each job/date pair. It’s kept separate from run ID and from when the observation arrived. A delayed message about attempt one must not overwrite the outcome of attempt two.

I chose a conservative rule: the greatest recorded attempt wins. If attempt 1 succeeds but attempt 2 fails or is still running, the gate blocks: the earlier success is invisible until the newer attempt resolves. Only when the most recent attempt succeeds does the date become ready. Another system might accept the first success instead. That’s a different contract, with its own tests.

A running observation may transition to one terminal result. Replaying that result is harmless. Changing an already terminal result is rejected; a repair must carry a new attempt ordinal.

This policy still depends on the collector. If the gate has not learned that a newer attempt exists, it cannot block on that invisible attempt. In a deployment, I would define who publishes authoritative observations and how missing or delayed updates are detected.

An executable failure, then a valid release

The lab uses two synthetic dependencies and a synthetic total of 17,500 integer cents. Its demo first records successful October 16 backfills with October 23 start timestamps. Checking October 23 returns false. Recording successful October 23 attempts makes that date ready.

Run these commands from the repository root:

python -m unittest discover -s python/business_date_gate -p 'test_*.py' -v
python python/business_date_gate/gate.py

The demo prints:

Backfill completed today: False
Required business date ready: True
First publication: True
Same publication retried: False

That last False is the retry being ignored: the existing publication stays and nothing gets written twice.

Put the output and its evidence in one transaction

A readiness check alone does not prevent two workers from publishing the same date. In this local example, the publication key is the consumer name plus business date.

The code opens a SQLite BEGIN IMMEDIATE transaction, reads readiness, writes the target total and stores the successful upstream run IDs. Both inserts commit together. A repeated call with the same key, dependency set and amount returns without inserting another row. A different amount or dependency set is rejected so a correction cannot silently masquerade as a retry.

After publication, an upstream failure does not erase the original evidence. Replaying the already committed result remains a no-op. Rebuilding or revoking that result would require an explicit repair workflow.

The failure test matters here. A SQLite trigger rejects the evidence insert on purpose, right after the target insert. The transaction rolls back both writes. Removing the trigger and retrying succeeds. A separate test launches two publishers on independent connections to the same temporary database; only one creates the publication.

All 25 new tests passed locally with Python 3.12 and SQLite. They also cover date validation, midnight crossings, missing jobs, delayed observations, state transitions and conflicting retries.

What the lab doesn’t cover

Output and evidence live in one SQLite database, so the atomic part holds there and nowhere else. Delta, S3 or a message broker need their own commit checks. The total is illustrative too; the gate doesn’t read upstream tables.

On Databricks, I’d pass business date as a job parameter and let a table-update trigger on a control table start the check. The trigger only decides when to look. The date and attempt rules above still decide if the output goes out.

Before deploying, I’d test the collector’s pagination, retries and repair handling, and map the real run states explicitly.

Code: https://github.com/jenniferevangelista-de/data-engineering-lab/tree/e3ddbf036417f64d4e2e8b1058df15780b935820/python/business_date_gate

References:

Databricks run-list filters: https://docs.databricks.com/api/jobs/v2/list-runs
Job parameters: https://docs.databricks.com/aws/en/jobs/job-parameters
Table-update triggers: https://docs.databricks.com/aws/en/jobs/trigger-table-update
SQLite transactions: https://www.sqlite.org/lang_transaction.html

A Freshness Check Before the 8 a.m. Report

Jennifer Evangelista — Wed, 21 Oct 2026 07:48:57 +0000

Check the output interval and event timestamps before the report becomes available to consumers.

The dashboard says the overnight pipeline succeeded. The BI report opened at 8:00 a.m. People are already discussing the morning numbers.

Someone from operations asks why yesterday’s transactions are still the newest records.

Pipeline finished successfully.

In this example, every expected task succeeded. The pipeline copied an old snapshot into a valid new file, and no task checked its event timestamps. I would put that check at the boundary where the report becomes available to consumers.

Three clocks, three different questions

Teams often compress distinct timestamps into a single idea of “latest”:

The run clock: when the orchestration system started and finished a run. This helps answer whether the workflow executed.

The interval clock: the logical period of data assigned to the run. In Airflow, a time-scheduled run has a data interval; its logical date commonly denotes that interval’s start, not the moment the tasks physically execute. A run for Monday’s data may legitimately execute after Monday has ended.

The event clock: the newest business event actually represented in the output. If an hourly orders dataset finishes at 8:00 but its newest order_created_at is 5:00, you have a freshness question even if the DAG status says success.

The three clocks can agree, disagree, or be incomparable depending on what the source produces. Never use a scheduler timestamp as a substitute for inspecting output events.

How a green DAG can deliver old data

Consider an hourly DAG that extracts orders, transforms them, and writes a report table. A source API is delayed but still returns a well-formed response. The transform accepts the response. The write succeeds. A terminal task sends the success notification. Yesterday’s data has completed today’s commute.

Nothing in that story proves that the output contains events within the business SLA.

There is another nuance: Airflow’s DAG-run status is derived from task states, including leaf tasks. Its documentation warns that certain trigger-rule arrangements can produce an apparently successful DAG run even when an intermediate task failed. A status light is useful, but it is not an end-to-end assertion about the delivered dataset.

This is why I do not recommend solving freshness incidents by simply adding more retries. Retrying a successful extraction of an old snapshot may give you the same old snapshot again.

Define freshness from the consumer backward

For this hourly orders example, I would agree on the following contract with the consumer:

The hourly orders dataset must contain at least one valid row, must be produced for the required interval, and must contain business event timestamps no older than 90 minutes at the 8:00 reporting checkpoint.

The 90-minute number is a made-up example SLA , not a recommendation for every company. For a dataset that receives no orders on quiet days, a mandatory recent event may be an invalid contract; use source heartbeat, completeness markers, or expected arrival windows instead. For batch snapshots, partition publication time may be more informative than event time. The right contract reflects producer behavior and consumer expectations.

Before marking a dataset ready, I want answers to five specific questions: Did the job publish the required data interval, or did it succeed while reprocessing an older one? Is the newest event recent enough for this consumer? Is the expected partition or count present, or are missing records being silently tolerated? Did the agreed validation checks pass before the table became visible downstream? Are timestamps timezone-aware, is event time distinct from ingestion time, and is clock skew bounded?

These are properties of the output and its contract, not properties of the colored dot next to a task.

Fail at the publication boundary

The companion lab contains a small Python contract that demonstrates the idea without requiring Airflow, Spark, or cloud credentials:

Open the freshness-contract lab and tests on GitHub .

The core function accepts a snapshot description, the current UTC instant, the required interval end, and a maximum event age. It returns specific breaches for an old interval, stale events, empty output, failed quality checks, or implausibly future-dated events. A strict wrapper raises an error so the publishing task can fail closed rather than completing with bad data.

For example, the synthetic case is:

from datetime import datetime, timedelta, timezone
from freshness_contract import Snapshot, assess_snapshot

t = datetime(2026, 10, 20, 8, tzinfo=timezone.utc)
snapshot = Snapshot(
    dataset="orders_hourly",
    dag_state="success",
    data_interval_end=t,
    latest_event_at=t - timedelta(hours=2),
    row_count=100,
    quality_checks_passed=True,
)

print(assess_snapshot(
    snapshot,
    now=t,
    required_interval_end=t,
    max_age=timedelta(minutes=90),
))
# ('event timestamp too old',)

The output reports event timestamp too old: the latest event is 120 minutes old against the example limit of 90 minutes, even though dag_state is success.

Run the standalone tests with Python 3.12+:

python -m unittest discover -s airflow/day_001_freshness_contract -p 'test_*.py' -v

The tests cover stale-but-green output, the wrong interval, missing timestamps, empty output, quality failures, invalid timezone handling, clock skew and a healthy dataset.

Scope: this is an original pure-Python illustration of a guard. It is not an installed Airflow operator and has not been exercised against an Airflow scheduler, Databricks, cloud storage, or production data. To deploy it, fetch real metadata from the produced dataset and invoke the strict check before marking the artifact ready for downstream use.

What about asset-aware orchestration?

Airflow 3 includes asset-aware scheduling, which is valuable when downstream workflows should respond to upstream asset events instead of an arbitrary timer. For task-declared outlets, Airflow marks an asset as updated when the producer task completes successfully. Externally emitted asset events need their own producer-side validation. Neither path inherently certifies every business-quality dimension of the data. Put the contract check before the event is emitted, or make consumers verify the contract independently.

Asset-aware scheduling solves when to trigger. Data contracts answer what is safe to consume. They complement each other; they are not substitutes.

The incident-review questions I would ask

Did the source publish new business events? What is the newest event timestamp and what does that field actually represent? Which interval did the job write? Can we distinguish a source outage from a legitimate zero-activity period? Was the output made visible only after validation passed? What alert goes to the owner when the contract is breached?

Those questions lead to an operational action. “Make the DAG green” usually leads to another green DAG.

If I had to choose one dashboard tile for an executive review, I would show data availability against the consumer’s agreed freshness contract , backed by evidence, alongside workflow status. The scheduler should remain visible. It should not be the sole source of truth.

A green DAG means the orchestration ran. A trustworthy dataset means the data met its promises.

References

AI-generated illustration of a successful workflow delivering stale data.

Three Orders In, Three Rows Out, 42.9% More Revenue

Jennifer Evangelista — Wed, 21 Oct 2026 07:34:51 +0000

An INNER JOIN can lose orders and multiply others in the same query. Here is a reproducible test for the failure that aggregate dashboards often miss.

AI-generated illustration of duplicate matches and an unmatched order. Synthetic amounts in integer cents.

A join can return the same number of rows as its input and still inflate revenue. In this synthetic example, three orders become three output rows, while the total rises from 17,500 to 25,000 cents: an increase of about 42.9%.

The row-count check approves. Accounting would like a word.

The aggregate count is unchanged because one duplicated order compensates for one missing order.

AI-generated comic about a row-count check that misses a revenue mismatch.

The join that cancels its own mistakes

We have three synthetic orders, in integer cents:

order customer amount_cents
o1 c1 10000
o2 c2 5000
o3 c3 2500

The source total is 17,500 cents. Our customer profile table has two historical versions for c1, one for c2, and no profile for c3.

SELECT o.order_id, o.amount_cents, p.tier
FROM orders AS o
INNER JOIN customer_profiles AS p
  ON p.customer_id = o.customer_id;

What happens?

o1 matches two profile rows, so its 10,000 cents are counted twice.
o2 matches once.
o3 disappears because its customer has no profile.

The output is still three rows, but the sum is now 25,000 cents. We lost one fact row and duplicated another. A row-count assertion looks reassuring precisely because the two errors compensate numerically. The query assumes one matching profile per customer, but the data contains two matches for one customer and none for another.

Define the grain before writing the join

The grain of the source is one row per order_id. The intended relationship is many orders to one current customer profile. But the customer_profiles table actually stores multiple versions for at least one customer.

An INNER JOIN only enforces its equality predicate. It does not invent a uniqueness constraint or choose “current” by intuition. A LEFT JOIN would preserve unmatched orders, but it would not prevent fanout when two profiles match.

Before joining, make the contract explicit:

Every order is associated with exactly one valid profile for the report’s business time.
The right side contains no more than one profile row per join key after applying that time or current-record rule.
An unmatched order is an exception to investigate, not something silently removed.
A revenue report reconciles both fact count and monetary totals in integer cents.

For a Type 2 dimension, the correct fix is normally a temporal rule, such as selecting the version effective when the order happened. Choosing MAX(profile_version) without understanding the business definition of “current” is not a neutral fix; it can rewrite history. If the contract truly needs the latest known state rather than an as-of state, document that choice.

Four checks that expose the issue

Check 1: duplicate dimension keys at the join grain.

SELECT customer_id, COUNT(*) AS matches
FROM customer_profiles
GROUP BY customer_id
HAVING COUNT(*) > 1;

This is intentionally conservative. In a history table, duplicates may be legitimate overall, but they are not legitimate in the dimension slice selected for a one-profile-per-customer join. Filter or window the history according to the correct as-of rule first, then check that derived slice.

Check 2: unmatched orders.

SELECT o.order_id
FROM orders AS o
WHERE NOT EXISTS (
  SELECT 1 FROM customer_profiles AS p
  WHERE p.customer_id = o.customer_id
);

Use this even if the row-count totals happen to match. If null business keys are allowed, specify their matching policy explicitly instead of relying on implicit SQL NULL behavior.

Check 3: reconcile totals before and after.

SELECT COUNT(*) AS row_count, SUM(amount_cents) AS cents FROM orders;

SELECT COUNT(*) AS row_count, SUM(o.amount_cents) AS cents
FROM orders AS o
INNER JOIN customer_profiles AS p
  ON p.customer_id = o.customer_id;

For this example the counts both equal three, but the amounts disagree: 17,500 versus 25,000 cents. Finance-grade pipelines should avoid floating-point money and define how refunds, cancellations, currency, and tax are represented.

Check 4: fail the contract, not only the SQL execution.

A query that returns rows has not proved its relationship assumptions. Treat fanout, unmatched facts, and unreconciled totals as observable contract breaches. Fail the publication gate before an incorrect result enters an executive dashboard.

A small executable proof

Spark join hints suggest execution strategies, such as broadcasting one side; they do not enforce one profile per customer. Keep the cardinality and reconciliation checks when porting this pattern, then evaluate the physical plan separately. The executable proof here remains SQLite-only.

Spark JOIN reference: https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-join.html

Join hints: https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-hints.html

The public lab includes a dependency-free SQLite example and eight tests:

https://github.com/jenniferevangelista-de/data-engineering-lab/tree/main/sql/day_002_join_cardinality

The example deliberately produces the deceptive equality of row counts, catches duplicate profile keys, catches orphan orders, compares totals, and refuses the publication contract. Tests also cover a healthy dimension, an empty input, and a refund represented as negative integer cents.

Run it with Python 3.12+:

python -m unittest discover -s sql/day_002_join_cardinality -p 'test_*.py' -v
python sql/day_002_join_cardinality/join_guard.py

Scope: the tests run against SQLite with synthetic data. They do not benchmark Spark, execute a Databricks job, or validate your production dimensional model. The behavioral principle is portable; query plans and enforcement mechanisms differ by engine.

What I would require in a production review

A code review focused only on syntax and task status will not catch this class of incident. I would ask for a written grain, a documented as-of join rule, checks for right-side cardinality and unmatched keys, monetary reconciliation, and a failure path that blocks consumption when those checks fail.

For an analytics system with late-arriving dimensions, the contract also needs a defined quarantine or remediation strategy. Do you defer the record, show it as “unknown,” or resolve it from an earlier version? A blanket INNER JOIN is not a policy.

For this example, I would block the report until c1 has one profile selected by the agreed time rule and o3 has an explicit resolution. Then I would rerun the key, row-count, and money checks before making the result available.

References

https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-qry-select-join (INNER vs LEFT and join criteria).
https://www.postgresql.org/docs/current/queries-table-expressions.html (relational join behavior).
https://github.com/jenniferevangelista-de/data-engineering-lab/tree/main/sql/day_002_join_cardinality (synthetic example and tests).

Unity Catalog Row Filters and Column Masks: A Fail-Closed Security Design You Can Test

Jennifer Evangelista — Mon, 19 Oct 2026 21:00:35 +0000

Original AI-generated illustration: column masking does not grant row access.

A data-governance slide says “only authorized analysts can see customer PII.” That sentence sounds reassuring until someone asks which identity was used for the test, whether null attributes are allowed through, and what happens when the same table is queried from a different notebook or SQL warehouse.

I prefer a claim that can fail a test. For a particular table, an unauthorized principal should see no disallowed rows and no raw protected values , even when the query is not the one the policy author anticipated. Databricks Unity Catalog row filters and column masks provide table-level controls at query time, but their reliability depends on function logic, type compatibility, privilege management and execution context. For policies that must span many tables, Databricks now recommends attribute-based access control (ABAC) instead of maintaining a function attachment by hand on every table.

Let’s build the smallest realistic demonstration and then attack our assumptions. A mask should protect the email column, not cosplay as an access-control strategy.

Row filtering and masking answer different questions

A row filter is a SQL user-defined function that returns a Boolean for each row. FALSE and NULL both mean the row is not returned. It is appropriate for conditions such as region, tenant, department or legal entity, once the identity source and authorization model are defined.

A column mask is a SQL UDF applied when a protected column is read. It returns a value with a type compatible with the original column. A mask can preserve a value for an authorized group and return a redacted placeholder for everyone else. One table can have a row filter and multiple column masks, but each masked column has one active mask.

The design rule is straightforward: a missing or unexpected condition should not broaden access. The SQL below uses an explicit allowlist for regions and a default-redacted mask.

Reproducible Databricks SQL lab

This lab requires a Unity Catalog-enabled Databricks workspace, SQL warehouse or supported Databricks Runtime, a writable catalog, and permission to create a schema, functions and a test table. Replace main with your authorized catalog. The group names are fictional and must exist in your account for the identity-specific test matrix. All customer data below is fake. Do not run this against a production table.

Run the following statements in a Databricks SQL editor. An account admin should provision three test groups: governance-admins, br-analysts and pii-readers. The testers must also receive the ordinary privileges necessary to query the demo table. These groups are demonstrations, not a suggested enterprise naming standard.

CREATE SCHEMA IF NOT EXISTS main.jta_governance_demo;

CREATE OR REPLACE FUNCTION main.jta_governance_demo.allow_region(region STRING)
RETURNS BOOLEAN
RETURN
  is_account_group_member('governance-admins')
  OR (region = 'BR' AND is_account_group_member('br-analysts'));

CREATE OR REPLACE FUNCTION main.jta_governance_demo.mask_email(raw_email STRING)
RETURNS STRING
RETURN CASE
  WHEN is_account_group_member('pii-readers') THEN raw_email
  ELSE 'REDACTED'
END;

CREATE OR REPLACE TABLE main.jta_governance_demo.orders (
  order_id BIGINT,
  region STRING,
  customer_email STRING
) USING DELTA;

INSERT INTO main.jta_governance_demo.orders VALUES
  (1, 'BR', 'ana@example.invalid'),
  (2, 'US', 'bob@example.invalid'),
  (3, NULL, 'unknown@example.invalid');

ALTER TABLE main.jta_governance_demo.orders
SET ROW FILTER main.jta_governance_demo.allow_region ON (region);

ALTER TABLE main.jta_governance_demo.orders
ALTER COLUMN customer_email
SET MASK main.jta_governance_demo.mask_email;

SELECT order_id, region, customer_email
FROM main.jta_governance_demo.orders
ORDER BY order_id;

What should this return? It depends on the querying principal. You cannot adequately validate the policy by rerunning SELECT as its owner and assuming every analyst has the same view.

Querying identity: Member of br-analysts only; Expected rows: Order 1; Expected email values: REDACTED

Querying identity: Member of br-analysts and pii-readers; Expected rows: Order 1; Expected email values: ana@example.invalid

Querying identity: Member of governance-admins but not pii-readers; Expected rows: Orders 1, 2, 3; Expected email values: All REDACTED

Querying identity: Member of none of these groups; Expected rows: No rows; Expected email values: No values

Querying identity: Member of pii-readers only; Expected rows: No rows; Expected email values: No values

Two details are intentional. First, the null region is not visible to a Brazil-only analyst because the allow condition is not true for that record. Second, membership in pii-readers does not grant row access on its own. The mask governs values of rows that survived the filter; it does not manufacture access to protected rows.

One nuance matters for the third row: NULL passed through allow_region is allowed only when the governance-admins branch is true. Everyone else sees it filtered out. That is a conscious administrative rule, and it should be reviewed as part of the access model.

Tests worth automating

A screenshot from the SQL editor is not a security test suite. I would maintain a test matrix with the expected counts and masked values for several dedicated non-production principals. Repeat it whenever group assignments, the UDFs or the table schema change.

At minimum, automate these acceptance checks:

A principal without any allowed groups gets zero rows. This tests the default-deny path.
A Brazil-only analyst cannot retrieve the US row, including by joining on a supposedly hidden identifier.
A PII-only principal does not gain region permissions by accident.
An analyst loses raw email access immediately after removal from pii-readers, subject to normal identity propagation and cache behavior in your environment.
A null or unexpected region does not appear to a region-limited analyst.
Query plans and performance remain acceptable with the filter and mask enabled.

A test harness can run SELECT COUNT(*) and SELECT customer_email as each principal and compare the result set to a fixed contract. Do not rely on a session variable that pretends to change the current account identity ; the authorization functions use the caller's real identity and group membership.

The most dangerous bug may be a data type, not a missing policy

Databricks documentation describes a subtle failure mode: if a table column and its UDF parameter have different types, implicit casts occur. With ANSI mode disabled, values that cannot be cast can become NULL silently. A filter written with an unfortunate IS NULL branch could unintentionally allow rows after such a cast.

Keep the UDF signature aligned with the actual column type, prefer predictable expressions, and validate policies with data containing nulls and malformed values. For sensitive workloads, follow the documented recommendation around ANSI behavior and type matching rather than trusting a policy simply because it compiled.

Also keep the masking function as simple as possible. More arguments, Python UDFs or mapping-table lookups can cost optimization opportunities. Filtering and masking are a security boundary: the engine may prioritize correct policy enforcement over aggressive pushdown, and that tradeoff can show up in query latency.

Table-level rules, ABAC and dynamic views are not interchangeable

Manual table-level UDF attachment is a good fit when there are few tables, well-defined owners and localized requirements. It is a weaker governance operating model when hundreds of independently managed tables need the same sensitive-column policy. A table owner or authorized manager can change a manual attachment, so your organization must audit changes and control policy administration.

ABAC policies are a stronger default for centrally managed rules: policy functions can be attached at catalog or schema scope and target tables/columns using governed tags. This helps new tagged tables inherit policies and separates stewardship from authoring. Databricks recommends ABAC for consistent filtering and masking across many tables; verify runtime and privilege requirements before adopting SQL CREATE POLICY in an older workspace.

Dynamic views are useful when consumers should access only a curated SQL interface, such as a joined or transformed subset of base tables. They are not the same as attaching a table-level rule to every direct reader of the base table. The choice should follow the threat model and how users are actually granted access.

None of these replaces clean catalog permissions. Users need the appropriate table and schema privileges in addition to whatever row-level or column-level policy logic applies. In other words, a mask is not permission to query everything, and a SELECT grant is not permission to ignore a mask.

Production limitations you should put in the architecture decision

Databricks lists important constraints for tables with row filters or column masks. Among them: path-based file access to protected tables is not a supported way to query through the policy; time travel and cloning have restrictions; some Delta APIs and interoperability surfaces are unsupported; certain UPDATE, DELETE and MERGE patterns have runtime-dependent limitations. Dedicated-access compute has additional version and serverless fine-grained-access-control requirements. These constraints are design inputs , not footnotes to discover after migration.

The example uses manual table-level policies and therefore should not be mistaken for a demonstration of centralized ABAC. The SQL syntax can be checked against documentation, but there is no claim here that the example has been executed in a Databricks workspace. Group creation, permissions, supported compute/runtime and behavior under each identity must be validated there before use in production.

For lab cleanup, detach the policies before dropping their functions:

ALTER TABLE main.jta_governance_demo.orders
  ALTER COLUMN customer_email DROP MASK;
ALTER TABLE main.jta_governance_demo.orders DROP ROW FILTER;
DROP TABLE main.jta_governance_demo.orders;
DROP FUNCTION main.jta_governance_demo.mask_email;
DROP FUNCTION main.jta_governance_demo.allow_region;
-- Drop the demo schema only if nothing else uses it.

Detaching first matters. Databricks explicitly warns that deleting a referenced masking function while leaving the mask attached can leave a table inaccessible until the orphaned reference is removed.

The takeaway

Good governance is a set of falsifiable expectations: which identity can see which row, and which value, under which compute path? Start with a small allowlist, design for missing attributes, test several principals, and write down what you have not tested. For organization-wide coverage, move beyond copy-paste table functions toward centrally governed ABAC and audit your permissions continuously.

A secure dashboard is not the number of policies created. It is the number of unauthorized paths you have proved do not work.

Official sources

Databricks, Row filters and column masks (overview, ABAC choice, compatibility and limits): https://docs.databricks.com/aws/en/data-governance/unity-catalog/filters-and-masks/
Databricks, Manually apply row filters and column masks (UDF syntax, SET ROW FILTER, SET MASK, removing policies): https://docs.databricks.com/aws/en/data-governance/unity-catalog/filters-and-masks/manually-apply
Databricks, ROW FILTER clause : https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-row-filter
Databricks, Column mask clause : https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-column-mask
Databricks, CREATE POLICY (ABAC): https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-create-policy

Continue in the lab

The Data Engineering Lab separates runnable examples from service-integration claims. Its Community lab demonstrates synthetic SQL checks and explicit NULL handling; it is a testing reference, not a Unity Catalog authorization test suite. The identity matrix in this article still needs an isolated Databricks workspace.

For the complementary data-correctness contract, read Your Delta MERGE Is Not a Mind Reader. Recovery and replay are covered in Your S3 Pipeline Needs a Replay Plan, Not More Optimism.

Your Kafka Consumer Lag Is Not Your SLA: Offset Math, Retention, and the Missing-Commit Trap

Jennifer Evangelista — Mon, 19 Oct 2026 20:58:58 +0000

Original AI-generated illustration: a zero-lag dashboard can hide a missing committed offset.

A dashboard declares that your Kafka consumer lag is zero. Support closes the incident. The business report is still missing records.

The chart is happy. The business would like its data back.

The apparent contradiction is not necessarily a Kafka defect. It may be a defect in the meaning assigned to the metric. Kafka offsets tell us how a particular reader relates to a particular partition’s log. They do not, on their own, certify that every business event was processed, that a downstream database commit succeeded, or that a latency target was met.

The distinction matters because the common formula, log_end_offset - committed_offset, is only meaningful after checking that both offsets exist, refer to the same topic-partition and are compatible with the log's retention boundary. Before summing anything across partitions, I want to know which measurements are trustworthy.

Three offsets, three different questions

For each topic-partition, collect these values:

Beginning offset (B). The earliest offset still available to read from the log. Retention and log compaction can affect the relationship between offsets and actual accessible records.
End offset (E). The boundary representing the next offset at the end of the readable log, according to the API and isolation level used by your measurement.
Committed consumer-group offset, C. The offset the group has recorded as its next position. A committed offset is not a receipt from your destination system.

When B <= C <= E, the familiar backlog estimate is:

partition_offset_lag = E - C

Example: if a group has committed offset 140 and the observed end is 150, the offset lag is 10. If C equals E, lag is zero. That is a specific statement about the observed positions, not proof of end-to-end correctness.

The assumption that breaks many homegrown monitors is that C is always present and always falls in that range. It isn’t.

Four statuses instead of one deceptive number

Consider an orders topic observed at 08:00. Four partitions show different states:

Partition: 0; B: 100; E: 150; C: 140; Correct interpretation: Offset lag = 10; position is within the retained range.

Partition: 1; B: 200; E: 260; C: null; Correct interpretation: No committed offset is available. Lag is unknown , not zero.

Partition: 2; B: 500; E: 620; C: 480; Correct interpretation: The committed position precedes retention by 20 offsets. Investigate potential unrecoverable gaps.

Partition: 3; B: 700; E: 740; C: 750; Correct interpretation: The committed position is beyond the observed end. Validate snapshot consistency and log history.

These become four operational states:

OK: B <= C <= E. Compute E - C.

NO_COMMITTED_OFFSET: C is absent. Do not substitute zero. This may be a new group, a deleted commit, a partition with no recorded group offset, or a mismatch between group and topic-partition inventory.

BEHIND_RETENTION: C < B. The span B - C is an unavailable offset range. It is not automatically the number of lost business events, especially on compacted topics, but it is a concrete reason to stop presenting the group as healthy.

AHEAD_OF_LOG_END: C > E. Never clamp a negative result to zero. A racing measurement, broker/log changes or an incorrect partition association could have produced inconsistent boundaries; investigate before claiming anything about throughput.

When even one partition is unknown or inconsistent, my group-level total_lag is null. A partial sum of healthy partitions can still be shown, but it must not be labeled the total.

A reproducible audit in plain Python

You can test these rules with synthetic snapshots without standing up Kafka. Save this as lag_demo.py and run python lag_demo.py with Python 3.10 or later:

from dataclasses import dataclass

@dataclass(frozen=True)
class Snapshot:
    partition: int
    beginning: int
    end: int
    committed: int | None

def classify(row: Snapshot) -> tuple[str, int | None]:
    b, e, c = row.beginning, row.end, row.committed
    if b < 0 or e < b:
        raise ValueError("invalid retained log boundary")
    if c is None:
        return ("NO_COMMITTED_OFFSET", None)
    if c < 0:
        raise ValueError("negative committed offset")
    if c < b:
        return ("BEHIND_RETENTION", None)
    if c > e:
        return ("AHEAD_OF_LOG_END", None)
    return ("OK", e - c)

snapshots = [
    Snapshot(0, 100, 150, 140),
    Snapshot(1, 200, 260, None),
    Snapshot(2, 500, 620, 480),
    Snapshot(3, 700, 740, 750),
]
results = [classify(row) for row in snapshots]
for row, (status, lag) in zip(snapshots, results):
    print(f"partition={row.partition}, status={status}, lag={lag}")

assert results == [
    ("OK", 10),
    ("NO_COMMITTED_OFFSET", None),
    ("BEHIND_RETENTION", None),
    ("AHEAD_OF_LOG_END", None),
]

Expected output:

partition=0, status=OK, lag=10
partition=1, status=NO_COMMITTED_OFFSET, lag=None
partition=2, status=BEHIND_RETENTION, lag=None
partition=3, status=AHEAD_OF_LOG_END, lag=None

This example classifies provided snapshots; it does not connect to a broker. A production integration would fetch committed group offsets through Kafka’s Admin API (listConsumerGroupOffsets), and the beginning/end boundaries through listOffsets, grouping by exact topic and partition. It would also attach group_id, collection timestamp, cluster identity and client/isolation configuration to each snapshot. Because these calls are not an atomic multi-partition snapshot, transient inconsistencies should trigger a refresh and diagnosis rather than automatic data-loss claims.

The collector and the calculation should be separate components. The code above is the complete local calculation demo for this article; the public lab does not yet contain a Kafka collector or broker integration. For the same unknown-versus-zero principle in a runnable Spark example, see optional operation metrics in my Community lab. That example reads synthetic metric maps, not Kafka offsets.

Why low lag can coexist with a broken pipeline

Commits can run ahead of downstream durability. A consumer might commit offsets and then fail while writing to its warehouse. After restart, the consumer group position can look healthy while the target lacks events. Whether this is possible in a particular architecture depends on commit ordering, destination guarantees and transaction design. Kafka’s at-least-once and exactly-once discussions are about specific processing boundaries, not a magic property of the number in a dashboard.

A group can be fast but stale. A source may not have received events for an hour. Its end offset does not move, so its consumer can show zero lag. Your business freshness SLO is still breached. Measure source freshness, event-time age and sink visibility separately.

One aggregate hides a hot partition. A group with 200 partitions can have modest total lag but one partition carrying the orders that matter. Keep per-partition maxima and age percentiles, not just sum and average.

Offsets are not a wall clock. E - C = 10,000 can mean seconds or hours depending on the arrival rate. Converting count to time by dividing by a short-term rate is an estimate that fails when traffic is bursty or paused.

Retention changes recovery choices. If C falls below B, the consumer may need a policy decision about offset reset or reprocessing from a separate durable source. An automatic reset-to-latest can make the lag chart beautiful while discarding the very history the incident team needs. Define this behavior explicitly and log the decision.

Compacted topics complicate counts. Offsets remain positions in the log, but not every offset within a numeric range necessarily represents a currently readable user record. Do not market offset differences as exact record counts for compacted workloads.

An on-call contract I would actually sign

A useful consumer dashboard has at least three panels. Position health shows per-partition B/C/E, group assignment, committed-offset availability and retention gaps. Pipeline health tracks successful writes, retries, rejected records, deduplication/transaction results and error budgets. Business freshness measures when the most recent valid business event became queryable in the serving layer, relative to the time it was expected.

Alerts should distinguish “backlog above threshold” from “cannot calculate lag,” “committed position outside retention,” and “data not visible downstream.” The first is often a capacity or rate problem; the others can be correctness incidents. Treating all four as the same red/green threshold destroys useful diagnosis.

When a pipeline is repaired, I would require a recovery test: record a known input key, trace its topic-partition-offset, verify the target effect, force a controlled retry, and confirm the intended duplicate-handling behavior. A lag graph cannot run this test for you.

Limitations and where to be careful

The Python example is a local calculation test with fabricated values. It does not validate authorization, Kafka brokers, consumer rebalances, committed-offset races, aborted transactions, compaction behavior or sink idempotency. Kafka’s Admin and consumer APIs differ in how they expose group state and offset boundaries; production collectors need a tested client/version policy. Do not infer that every missing committed offset means data loss, or that every retention gap equals lost events.

If you use Structured Streaming with Kafka, its checkpointed progress and sink semantics must be analyzed as a separate implementation, not conflated with an arbitrary Kafka consumer group’s commits. That is another reason I keep source recovery, pipeline state and business outcomes as separate evidence.

The takeaway

Offset lag is useful once you have proved that the inputs make sense. It tells you how far a recorded position is from an observed log boundary. It doesn’t answer whether the pipeline fulfilled its contract. Keep unknowns unknown, surface retention boundaries, and make the business SLO visible on its own terms.

Official sources

Apache Kafka, Admin API and consumer API; consumer groups and delivery semantics.

Continue in the lab

Data Engineering Lab contains small reproducible examples and explicit validation limits. CDC preparation covers replay, ordering and conflicting versions with synthetic data; it does not validate a live Kafka sink.

For recovery design, read Your S3 Pipeline Needs a Replay Plan, Not More Optimism. For deterministic target updates, read Your Delta MERGE Is Not a Mind Reader.

Your S3 Pipeline Needs a Replay Plan, Not More Optimism

Jennifer Evangelista — Sun, 18 Oct 2026 11:06:23 +0000

How file checkpoints, event identity, and source versions work together in a Databricks ingestion design on AWS.

A file lands in S3. The job turns green. Everyone enjoys a brief, suspiciously peaceful moment.

Then a recovery process delivers the same business events in a different file. The ingestion reads both objects correctly, and the downstream table now has duplicate entities or an older customer address. The green indicator has done its job: it reported that the process finished. Unfortunately, nobody had asked it to settle questions of identity and history.

Reliable replay starts by giving each layer a smaller, clearer responsibility. A file checkpoint, an event identifier, and a source version solve different problems.

This reference design includes a sandbox configuration and a recovery rehearsal plan.

This article assumes familiarity with Auto Loader, Delta Lake, and structured streaming on Databricks. If you are new to those concepts, the Auto Loader overview and Delta Lake documentation are good starting points before continuing.

There are three different things to recognize

First, identify the physical input. Which object did the reader discover and process? File-level identity and ingestion metadata help answer that question and support recovery of the reading process.

Second, identify the logical event. Two different objects can carry the same source event. Reading each object once does not establish that each business event appears once.

Third, identify precedence for an entity. A customer can have many legitimate events. A current-state table must know which version takes precedence, while a history table may need to retain every meaningful transition.

Write down the contract before choosing a deduplication expression. For each layer, name the identity, the ordering rule, the treatment of conflict, and the recovery behavior. “Exactly once” is more informative when followed by an explanation of exactly what is counted once.

Give the layers concrete responsibilities

A small reference architecture might look like this:

Producer -> S3 -> Auto Loader -> Bronze Delta
                                     |
                           validation / quarantine
                                     |
                         Silver state or history
                                     |
                          contract-based consumption

Bronze preserves received data and provenance. Silver applies the business interpretation: keys, versions, operations, and the destination model. Consumers receive a documented view rather than inheriting every ingestion assumption.

These layer names are convenient, not mandatory. The useful design decisions are who owns each output, how long it is retained, how it is rebuilt, and which failures stop a batch from being promoted.

S3 notifications can be delivered more than once, and their arrival does not establish a global business order. Treat a notification as a discovery signal, not as a reliable “this customer version is newest” certificate. Amazon S3 documentation.

A checkpoint tracks the reader’s progress

Auto Loader uses checkpoint-associated state to track discovered files. Its supported file-ingestion guarantees do not turn repeated events inside different files into unique business events. Auto Loader overview.

Here is an illustrative sandbox configuration:

from pyspark.sql import functions as F

# Replace these with authorized sandbox locations and identifiers.
source_path = "s3://YOUR_BUCKET/customer-events/"
schema_location = "s3://YOUR_STATE_BUCKET/customer-schema/"
checkpoint_location = "s3://YOUR_STATE_BUCKET/customer-checkpoint/"
bronze_table = "sandbox.ingestion.customer_events"

incoming = (
    spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", schema_location)
    .option("cloudFiles.schemaEvolutionMode", "none")
    .option("rescuedDataColumn", "_rescued_data")
    .schema("customer_id STRING, source_version BIGINT, value STRING")
    .load(source_path)
    .select(
        "*",
        F.col("_metadata.file_path").alias("source_file"),
        F.col("_metadata.file_modification_time").alias("source_file_mtime"),
        F.current_timestamp().alias("ingested_at"),
    )
)

query = (
    incoming.writeStream
    .option("checkpointLocation", checkpoint_location)
    .trigger(availableNow=True)
    .toTable(bronze_table)
)
query.awaitTermination()

The selected metadata fields record the source file and its modification time alongside the ingestion timestamp. This makes it possible to trace the same business event arriving in two different objects. These are provenance fields, not a source-version ordering rule. Reserve these output column names in the ingestion contract. File metadata reference.

With a supplied schema, Auto Loader defaults to schema evolution mode none. Without a rescued data column, unexpected fields are ignored. The explicit rescuedDataColumn option above retains them for review instead of silently losing them. It does not add them as typed business columns or automatically quarantine malformed JSON. Schema and rescued data reference.

Use a dedicated checkpoint for each ingestion workload and preserve it during ordinary recovery. Without its checkpoint, Auto Loader starts without its previous file-processing history. With the default cloudFiles.includeExistingFiles=true, existing eligible files under the source path are included again, which can append their rows to Bronze a second time. Source filters and file availability still apply; this cannot recover files already removed. Auto Loader options.

Watch checkpoint size as the workload grows; state accumulates. The production docs cover the operational details that go beyond this configuration.Auto Loader production guidance.

Deleting a bookmark does not make a book shorter. It mainly makes remembering your place more exciting.

Business replay belongs in the destination contract

Imagine that customer A is already stored at source version 10. A replay brings version 8. The ingestion may have done everything correctly. Silver must still prevent the state from moving backward.

A possible current-state contract is:

Higher source version: eligible to replace the older state.
Same version, same payload: exact retry.
Same version, different payload: conflict requiring a defined resolution path.
Lower source version: older state; do not overwrite the newer state.

The complete implementation must enforce this contract across both the incoming batch and the existing target. It also needs explicit delete and concurrency rules; naming a column customer_id does not enforce uniqueness.

The PySpark lab demonstrates one part of the problem: validating a finite batch, collapsing exact retries, rejecting internal conflicts, and selecting the highest version per normalized key. I covered that part in Your Delta MERGE Is Not a Mind Reader.

For a history requirement, keep the distinction sharp. Reducing a batch to its latest event can be correct for a current-state table and wrong for a consumer that needs the intermediate changes.

Original AI-generated editorial illustration. The returning delivery is a metaphor for replay, not an AWS service diagram.

When your ingestion design already handles file checkpoints and event identity, the next question is whether managed abstractions add anything. The short answer: they handle sequencing and SCD mechanics, but not the business specification that determines what a meaningful change is.

Managed CDC still needs a business specification

Lakeflow’s AUTO CDC APIs offer source sequencing and SCD processing. Their usefulness does not remove the need to define keys, meaningful change columns, delete behavior, and the desired history model. CVerify that these APIs are available in your runtime before building on them.AUTO CDC reference.

For example, a technical ingestion timestamp should not create a new business version merely because the same event was delivered again. Conversely, a real change must not disappear because the comparison excluded an attribute the consumer cares about.

Review a small set of explicit examples with the producer and consumer. Specifications become much easier to evaluate when someone can point at two contradictory records rather than another paragraph containing the word “robust.”

Quarantine needs an exit strategy

Moving a bad row into a quarantine table is the beginning of resolution. Record the reason, a safe reference to the source, the rule version, and enough identity to connect the issue to its batch. Keep unnecessary personal data out of logs.

Then define the return path. Does a correction arrive with a new source version? Is the original event amended? Who approves a replay, and how is the destination protected from duplication?

Without that path, quarantine becomes a very well-organized place where data goes to wait indefinitely.

Do not resolve a conflicting version by picking the alphabetically largest payload merely to keep the job green. That choice is a business rule in disguise. If no valid precedence rule exists, stopping with an actionable explanation can be more useful than quietly storing an arbitrary answer.

Rehearse recovery with a small, unfriendly dataset

Start with synthetic data: two customers, multiple versions of one customer, and an exact retry. Check the expected bronze records and silver state. Record input rows, distinct identities, applied versions, and rejected conflicts.

Restart with the same checkpoint and inspect what actually happens. Deliver the same event in another object. Include an older version and a conflicting equal version. The target should follow the documented rules, and the conflict should reach the defined resolution path.

Also test an interruption at a relevant point in your chosen write mechanism. If the pipeline writes several tables, do not presume that those writes form one atomic transaction. Recovery needs to explain how an incomplete run is detected and reconciled.

Keep evidence from each rehearsal: input files, run identifiers, checkpoint used, expected and actual rows, and the reason each assertion passed or failed. A local transformation test, a service integration test, and an end-to-end recovery exercise answer different questions.

A green job should still have something useful to say

A process can finish successfully after receiving no fresh input. Monitor source freshness, processing delay, rejected records, and the consumer’s actual availability requirement alongside job completion.

Every alert needs an owner and a documented response path. Decide how to distinguish an upstream delivery failure from a transformation defect. The right freshness threshold comes from the use case, not from a universal number borrowed from somebody else’s dashboard.

The design holds up when you can explain, without looking at the code, what happens to a repeated event, an older version, and a conflict. Give those cases names and expected outcomes before they appear during an incident.

Your future on-call self may never send a thank-you note. A quiet pager is a perfectly respectable substitute.

Scope: a reference design with synthetic examples. The companion batch-preparation lab was validated on local Spark 3.5.3. The Auto Loader configuration and recovery rehearsal were not executed against AWS or a Databricks workspace. Running the configuration requires authorized S3 and state-storage access, a compatible runtime, and catalog/schema and target-table permissions. Malformed input, deletes, concurrent writers and full history need their own implementation and tests.

References checked on October 18, 2026.

Your Delta MERGE Is Not a Mind Reader

Jennifer Evangelista — Sun, 18 Oct 2026 07:42:44 +0000

A practical PySpark contract for duplicate events, conflicting versions, and deterministic updates.

Three source rows. One target row. Everyone insists they are the right match. Spider-Man pointing meme; image source: UOL / reproduction.

Three records walk into a MERGE. All claim to represent the same customer. One carries yesterday’s address. Two carry today’s address. The target table has one row, and Delta can refuse to update it from more than one matching source row, even when those rows agree.

If you have seen DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE, this article is for you.

The engineering problem is less about persuading Delta to accept the input and more about defining what the input means. Which records describe the same entity? What makes a version newer? When is a duplicate harmless, and when is it evidence that the producer has contradicted itself?

Those decisions belong in a data contract. A successful job is a poor substitute for one.

First, decide what “the same customer” means

Consider this synthetic batch:

key | sequence | value
" a " | 1 | old
"A" | 2 | new
"A" | 2 | new

For this example, the business considers letter case and leading or trailing spaces irrelevant. We therefore normalize the key with upper(trim(key)). All three records identify A.

This is a domain assumption, not a universal cleaning rule. If abc and ABC are different product codes in your system, converting them to uppercase would erase a meaningful distinction. Also, Spark's ordinary trim is not a complete Unicode normalization strategy. Be precise about the spaces and characters your contract accepts.

Apply the same identity rule everywhere it matters: validation, aggregation, deduplication, and the eventual match condition. Grouping original keys and normalizing only inside ON can produce several supposedly unique groups that collapse into one match key.

The records did not become duplicates because MERGE was feeling dramatic. The transformation exposed an identity rule that the earlier grouping had ignored.

“Latest” is a business rule with a sorting expression

In this lab, sequence is a non-negative BIGINT supplied by the producer. It orders versions within each normalized key. Version 2 takes precedence over version 1, regardless of arrival order.

Ingestion time is not automatically a safe substitute for a source sequence. A delayed version can arrive after a newer one. Sorting by arrival would then give the old state an undeserved promotion.

If your source uses a timestamp, define what happens when timestamps tie. A source sequence can resolve the tie if it represents a meaningful ordering. A random identifier may make the result repeatable, but repeatable does not necessarily mean correct.

For this example, the rule is deliberately strict: the same key and sequence must describe the same value. Two different values at the same position in the source history are a conflict. We reject the batch instead of inventing a winner.

Separate an exact retry from a contradiction

The implementation follows four steps:

Validate the required types and reject missing or blank keys and invalid sequences.
Normalize the key according to the lab’s identity rule.
Collapse exact repeats of key, sequence, and value, then reject multiple values for one key and sequence.
Select the highest sequence for each key.

The distinction matters. dropDuplicates(['key']) expresses which columns define duplication; it does not express a newest-version policy. Spark documents deduplication separately from the ordering you must define for your business rule. PySpark reference.

After conflict validation, the selection itself is small:

from pyspark.sql import Window, functions as F

order = Window.partitionBy("key").orderBy(F.col("sequence").desc())
prepared = (
    unique.withColumn("_rank", F.row_number().over(order))
          .where("_rank = 1")
          .drop("_rank")
)

Here, unique means the already-normalized and validated batch. Without that validation, row_number would silently pick one of the conflicting rows. A window function is very good at applying a rule; it cannot supply the missing business decision.

Add a fourth row, A | 2 | other, and the batch is rejected: same key, same sequence, different value.

The complete implementation is available in the Data Engineering Lab, including the checks around this selection.

Exact retry (same key, sequence, value): collapsed.
Same key and sequence, different value: batch rejected.
Higher sequence within the batch: wins.
Older than target: ignored by the illustrated MERGE below.

Exact retries can collapse. Contradictory versions need a decision. Original AI-generated editorial illustration.

NULL deserves an explicit policy too

The lab permits a NULL payload and does not interpret it as a delete. That choice is intentional. Some systems use null to mean an absent attribute, others use it to mean a cleared value, and others encode deletions in a separate operation field.

A NULL value and a text value at the same key and sequence still conflict here. The implementation removes exact duplicate rows, then counts distinct rows per key and sequence. That avoids relying on an aggregate over distinct payload values that could ignore NULL.

An empty DataFrame returns an empty result with the expected schema. A streaming DataFrame is rejected: the function accepts a finite batch and performs Spark actions during validation.

That last detail has an operational consequence. Use a stable input snapshot. If the underlying source can change between validation and evaluation, the function can validate one view of the input and later evaluate another. The contract assumes the batch remains the same batch.

Run the example without a cloud account

Clone the repository and run these commands from its root with Python 3.12 and Java 17 available:

python -m pip install -r pyspark/day_001_cdc/requirements.txt
python -m unittest discover -s pyspark/day_001_cdc -p 'test_*.py' -v

PySpark is pinned to 3.5.3. On Windows, the README explains the local Java and networking settings used for validation.

Eight tests pass against a real local Spark JVM. They cover normalized keys, exact retries, conflicting versions, NULL handling, invalid inputs, empty input, repartitioning, and idempotence of the preparation.

A well-prepared batch can still overwrite the wrong state

Suppose the target already contains version 10, and a later run receives version 8. The batch preparation correctly selects version 8 from that batch. The write must still prevent the persisted state from moving backward.

An illustrative Delta adaptation is:

MERGE INTO sandbox.customer_current AS t
USING prepared_updates AS s ON t.key = s.key
WHEN MATCHED AND s.sequence > t.sequence THEN
  UPDATE SET t.sequence = s.sequence, t.value = s.value
WHEN NOT MATCHED THEN
  INSERT (key, sequence, value)
  VALUES (s.key, s.sequence, s.value);

This assumes a target with unique normalized keys and non-null sequence values. Equal versions with conflicting source and target payloads require an additional validation policy. The lab detects conflicts inside a batch, not between that batch and an already-persisted target.

Runtime details matter as well. In DBR 15.4 LTS and below, duplicate-match detection considers the ON condition. In DBR 16.0 and above, it also considers the WHEN MATCHED conditions. Record your runtime and reproduce the actual case there. MERGE documentation.

Preserve history when history is the requirement

Keeping the last event per key is appropriate for some current-state workloads. It is not a general CDC algorithm. If a consumer needs every intermediate transition, discarding those transitions loses information even when the final state is correct.

Before promoting this pattern, test a repeated batch, an older incoming version, a conflicting version already in the target, and the intended delete behavior. Check the target’s uniqueness assumption and your concurrency model. Measure shuffle and validation costs using representative data rather than treating a small correctness test as a scale claim.

The useful question is not “How do I make MERGE stop complaining?” It is “Can I explain exactly which record should win, and why?”

Once that answer is explicit, the code has a job it can actually perform. Mind reading can remain outside the service-level agreement.

Scope: synthetic data, validated on local Spark 3.5.3. The MERGE example is illustrative and was not run against Delta. It does not cover deletes, concurrent writers or SCD Type 2 history.

References checked on October 18, 2026.