Turn an LLM into a working data-quality copilot on a deliberately dirty IoT dataset: profile it, have AI propose a test suite from the profile, prune the over-fitted rules the AI will inevitably include, and then triage a seeded two-part incident (a volume drop and a freshness stall) end-to-end — from ranked hypotheses to confirming SQL to a written RCA.
The dataset carries chronic dirt (duplicate keys, NULL devices, impossible readings, future timestamps, unicode names) plus an acute incident hidden in the last two days. The two skills this trains are the ones observability vendors now sell as products: generating quality rules you can actually keep, and compressing incident triage from an hour of ad-hoc queries to minutes of hypothesis-driven confirmation.
~120 minutes hands-on · Core tier (maps to module M3 · AI for Data Quality & Incident Response).
Local only — DuckDB plus Python, $0. Optionally the Anthropic Python SDK for the structured-triage step (a rules-based fallback is included).
mkdir dq-lab && cd dq-lab
python3 -m venv .venv && source .venv/bin/activate
pip install duckdb anthropic pydantic
Save make_dirty_data.py — 14 days of gateway sensor readings with chronic dirt every day and an acute incident on days 13–14:
"""Deliberately dirty IoT readings + a seeded volume/freshness incident on the last two days."""
import csv, random, os
random.seed(11)
OUT = os.path.dirname(os.path.abspath(__file__))
GATEWAYS = ["GW-01", "GW-02", "GW-03", "GW-04"]
rows, rid = [], 0
for day in range(1, 15): # 2025-07-01 .. 2025-07-14
for gw in GATEWAYS:
if day == 13 and gw == "GW-03":
continue # INCIDENT 1: GW-03 sends nothing on the 13th
n = random.randint(480, 520)
for _ in range(n):
rid += 1
hour = random.randint(0, 23)
if day == 14 and hour > 9:
continue # INCIDENT 2: nothing lands after 09:xx on the 14th
ts = f"2025-07-{day:02d} {hour:02d}:{random.randint(0,59):02d}:00"
device = f"dev-{gw[-2:]}-{random.randint(1,40):03d}"
temp = round(random.gauss(24, 4), 1)
hum = round(random.uniform(20, 95), 1)
r = random.random()
if r < 0.004: device = "" # NULL device
elif r < 0.006: temp = round(random.uniform(-90, -70), 1) # impossible temps
elif r < 0.008: hum = round(random.uniform(101, 130), 1) # humidity > 100
elif r < 0.009: ts = f"2026-07-{day:02d} 00:00:00" # future timestamp
elif r < 0.010: device = "dev-ñ-" + device # unicode name
rows.append([rid, device, gw, ts, temp, hum])
if random.random() < 0.002: # duplicate reading_id
rows.append([rid, device, gw, ts, temp, hum])
with open(f"{OUT}/iot_readings.csv", "w", newline="") as f:
w = csv.writer(f); w.writerow(["reading_id", "device_id", "gateway", "read_at", "temp_c", "humidity_pct"])
w.writerows(rows)
print("total rows:", len(rows))
Run python make_dirty_data.py — expected: total rows: 26422. Load it once into DuckDB (duckdb quality.duckdb):
create or replace table iot_readings as
select * from read_csv_auto('iot_readings.csv', types={'read_at': 'TIMESTAMP'});
The scenario for the incident half of the lab: it is the afternoon of 2025-07-14 and a dashboard consumer has pinged you that "the sensor numbers look low today."
Profile before you propose. Rules generated from schema alone are guesses; rules generated from a profile are hypotheses. Run SUMMARIZE iot_readings; in DuckDB and capture the output, then add the two profile views SUMMARIZE cannot give you:
-- chronic-dirt counts
select count(*) as total_rows,
count(distinct reading_id) as distinct_ids,
sum(case when device_id is null or device_id = '' then 1 else 0 end) as null_device,
sum(case when temp_c < -60 then 1 else 0 end) as impossible_temp,
sum(case when humidity_pct > 100 then 1 else 0 end) as impossible_humidity,
sum(case when read_at > timestamp '2025-07-15' then 1 else 0 end) as future_ts
from iot_readings;
-- daily volume
select strftime(read_at, '%Y-%m-%d') as d, count(*) as n
from iot_readings where read_at < timestamp '2026-01-01'
group by 1 order by 1;
Expected: 26,422 total vs 26,371 distinct ids (51 duplicated rows), 126 blank devices, 42 impossible temperatures, 48 humidity readings above 100, 27 future timestamps. The daily series runs ~1,900–2,050 rows per day, then 1,479 on 2025-07-13 and 806 on 2025-07-14. Save all of it to profile.md — this is the context you will hand the LLM.
Have AI propose the suite. Paste profile.md plus the table schema into your assistant with: "Propose a data-quality test suite for this table as dbt-style tests and plain-SQL checks. For each rule state: what it asserts, the threshold, and why the profile justifies it. Cover keys, nulls, ranges, timeliness, and volume." You will get something in the 12–20 rule range. Save the raw output as proposed_rules.md — do not edit it yet.
Prune the over-fitted rules. This is the judgment step the lab exists for. Walk every proposed rule and sort it into keep / loosen / reject, writing one line of reasoning each. Patterns to hunt: thresholds photocopied from the profile (if the profile showed max(temp_c) = 41.2 the AI will propose temp_c <= 41.2, which pages you on the first hot day — the physical bound is the spec, not the sample); ranges that enshrine today's dirt (humidity_pct <= 130 because that was the observed max — the rule must be <= 100); uniqueness proposed on columns that are legitimately non-unique (device_id); and volume rules pinned to exact daily counts instead of a tolerance band (right form: day-over-day drop > 30% alerts). A defensible pruned suite lands at 8–10 rules. Save as test_suite.sql with each rule as a query returning violating rows.
Codify the volume and freshness monitors. These two are the incident detectors, so write them yourself rather than trusting the draft:
-- volume: today vs trailing-7-day median, alert under 0.7
with daily as (
select strftime(read_at, '%Y-%m-%d') d, count(*) n
from iot_readings where read_at < timestamp '2026-01-01' group by 1)
select d, n,
round(n / median(n) over (order by d rows between 7 preceding and 1 preceding), 2) as vs_median
from daily order by d;
-- freshness: minutes since last accepted record
select max(read_at) as last_record,
date_diff('minute', max(read_at), timestamp '2025-07-14 18:00:00') as minutes_stale
from iot_readings where read_at < timestamp '2026-01-01';
The volume monitor flags 2025-07-13 (~0.73) and 2025-07-14 (~0.41); freshness shows the last record at 09:59 — 481 minutes stale at the 18:00 "now". Two independent alarms, possibly one incident, possibly two.
Triage with the copilot — structured, not freeform. Feed the LLM exactly what an on-call engineer would have: the two alert readouts, the daily volume series, and the schema. Ask for ranked hypotheses with confirming queries, as structured output so the result drops straight into your incident tooling. With the Anthropic SDK:
import anthropic
from pydantic import BaseModel
class Hypothesis(BaseModel):
rank: int
cause: str
reasoning: str
confirming_sql: str
class Triage(BaseModel):
incident_summary: str
hypotheses: list[Hypothesis]
client = anthropic.Anthropic()
response = client.messages.parse(
model="claude-opus-4-8",
max_tokens=16000,
thinking={"type": "adaptive"},
messages=[{"role": "user", "content": f"""You are a data-incident triage copilot.
Table: iot_readings(reading_id, device_id, gateway, read_at, temp_c, humidity_pct).
Sources: 4 gateways (GW-01..GW-04) stream readings continuously; a loader lands them hourly.
Alerts at 2025-07-14 18:00:
- Volume: 2025-07-13 at 0.73x trailing median; 2025-07-14 at 0.41x and falling.
- Freshness: last accepted record 2025-07-14 09:59 (481 min stale).
Daily volumes: {open('profile.md').read()[:2000]}
Rank the most likely root causes (max 5). Each needs one confirming DuckDB query
that would isolate it (slice by gateway, by hour, by load batch)."""}],
output_format=Triage,
)
triage = response.parsed_output
for h in triage.hypotheses:
print(h.rank, h.cause, "\n ", h.confirming_sql)
No API key available? Write the ranked list yourself from the same prompt material — the discipline (hypotheses before queries) is the lesson; the LLM is the accelerator.
Confirm or kill each hypothesis. Run the confirming queries top-ranked first. The two that decide the incident:
-- slice the volume drop by gateway
select gateway, count(*) n from iot_readings
where strftime(read_at, '%Y-%m-%d') = '2025-07-13' group by 1 order by 1;
-- slice the stall by hour on the 14th
select strftime(read_at, '%H') h, count(*) n from iot_readings
where strftime(read_at, '%Y-%m-%d') = '2025-07-14' group by 1 order by 1;
Query 1 returns only GW-01, GW-02, GW-04 (≈481/481/517) — GW-03 is entirely absent on the 13th while the other gateways are normal: a source-side outage, not a pipeline problem. Query 2 shows all four gateways present but every hour after 09 empty — all sources alive, nothing landing: a load-side stall. Two distinct incidents that co-fired one dashboard complaint. Kill the hypotheses the data rules out (seasonality, dedup misfire, partial-file load) explicitly.
Write the RCA. incident_rca.md, four sections: Impact (which days/consumers, quantified: −27% volume on the 13th, feed frozen since 09:59 on the 14th); Root causes (two, each with its confirming query and result pasted in); Ruled out (each killed hypothesis and the query that killed it); Follow-ups (per-gateway volume monitors so a single-gateway outage cannot hide inside a total, freshness SLA on the loader, and the pruned test suite from step 3 scheduled daily). Have the LLM draft the consumer-facing two-paragraph summary from your RCA — then fact-check every number in it against your queries before sending; the copilot drafts, the engineer owns accuracy.
Profiles make AI-generated rules testable, pruning makes them keepable, and slicing (by gateway, by hour) is what turns an alert into a root cause — the LLM accelerates every step but signs off on none of them.
profile.md contains the SUMMARIZE output, the chronic-dirt counts (51 / 126 / 42 / 48 / 27), and the daily volume series showing 1,479 and 806 on the last two days.proposed_rules.md (raw AI draft) and test_suite.sql (pruned) both exist, with a written keep/loosen/reject verdict per rule and 8–10 survivors.incident_rca.md identifies two separate root causes (GW-03 source outage on the 13th; loader stall after 09:59 on the 14th), each with its confirming query, and explicitly rules out at least two alternatives.The dq-lab/ folder: profile, raw and pruned rule suites, monitor queries, triage transcript (LLM or manual), and the RCA doc. Self-grading rubric (pass = 8/10):
random.seed(11) must be set before any other random call; regenerate the CSV.read_at as VARCHAR — reload with the types argument shown in Setup.1 preceding.messages.parse errors: check pydantic is installed and you are passing the class itself (not an instance) as output_format.Profiles and alert readouts usually leak less than raw data, but they still contain schema names, volumes, and business patterns — treat them as client-confidential. Route the rule-proposal and triage prompts through the client's approved tenancy (Amazon Bedrock or Azure OpenAI) or a local model via Ollama; the profiling SQL, monitors, and confirming queries are all local DuckDB and never touch an LLM. Where even schema names are restricted, profile with neutralized column aliases (col_a, col_b) and keep the mapping file inside the client boundary. The deliverable that matters — the RCA with confirming queries — is fully reproducible by the client without any AI access.
reading_id rows; 126 blank device_id; 42 temperatures below −60 °C; 48 humidity readings above 100; 27 timestamps a year in the future; a handful of unicode device names (valid data — a keep-as-is finding, not a rule).temp_c <= 41.2 (sample max as threshold), humidity_pct <= 130 (enshrines the dirt), uniqueness on device_id, and volume pinned to ~2,000 rows/day instead of a relative band.