Real-Time Anomaly Detection

A field of weather sensors reports wind, rain and temperature around the clock. Almost every reading is unremarkable, and there are far too many to look at — the historical file loaded below holds about a million of them, with new ones arriving continuously. The readings worth an alert are the few that break the pattern: a wind value that spikes while nothing else moves, a temperature that shifts further in an hour than it usually does in a month. You cannot write a rule for every sensor in advance, and you cannot eyeball a million rows. This notebook builds something that decides for you, in SQL and Python, with no trained model.

The idea: turn each reading of wind, rain and temperature into a vector, then label a new reading by finding the readings it most resembles and taking the label they already carry. Similar readings get similar vectors, so "unusual" becomes a distance you can compute — and computing it is a dot_product() query, so it happens inside the database, next to the data.

What you will do:

1. Load a million labelled historical readings and a stream of new ones from S3 using pipelines, SingleStore's built-in ingest.

2. Turn the new readings into vectors with UMAP.

3. Score each new reading against the whole history with dot_product(), let its twenty nearest neighbours vote on a label, and write the result back.

4. Watch the result in two dashboards — anomaly trends over time with Plotly, and a live table that refreshes as rows land.

No prior experience with vectors is needed. If you have used a SELECT, you can follow the scoring query.

Database Setup

Two tables hold everything in this notebook:

• sensor_data_with_vectors — the history you score against. Each row is one reading: date, city, longitude, latitude, the three sensor values, the anomaly label the reading carries, and the embeddings vector that represents it.

• sensor_data_stage — the landing table for new, unlabelled readings. Same columns, no anomaly.

The source dataset is French, so the sensor columns are named vent, pluie and temp — wind, rain and temperature. The anomaly column reuses those same three words to say which channel broke the pattern in that reading, or none if nothing did. The column names are left as they are so they match the CSV files on S3, but the charts and the live dashboard later on display them in English.

The cell below creates the database. A Free Starter Workspace already has one and does not allow creating another, so it checks is_shared_tier first and only runs DROP DATABASE / CREATE DATABASE on a Standard or Premium Workspace. On a Free Starter Workspace the tables are created in the database you already have.

Loading the Data with Pipelines

Two pipelines feed the tables, both reading CSV from a public S3 bucket:

• sensor_data_pipeline → sensor_data_with_vectors, the roughly one million historical readings that already carry a label and an embedding.

• sensor_realtime_data_pipeline → sensor_data_stage, the new readings to be labelled. It sets date = NOW() as the rows arrive, so they look like they just happened.

Why this is a pipeline and not a load script

If you are coming from another database, this step is usually the one that lives outside it. You write a loader — a script, a scheduled job, a managed ingest service — that reads the bucket, parses the rows, and issues inserts. It is a separate thing to deploy, secure, schedule, monitor and pay for.

A pipeline is a database object instead. You create it with DDL, exactly like a table, and the cluster itself runs it. It is stored in the database, included in backups, and survives a restart. What makes that fast:

• Every partition pulls its own data. Nothing is funneled through a single coordinator. Each partition connects to the source, claims its own slice, parses it, and writes its local rows — so throughput scales with partitions rather than with one loader process.

• There is no intermediate copy. Rows go from the source straight into the target table.

• Progress is tracked transactionally. Committed files and offsets are recorded in the same transaction as the data, so a stopped pipeline resumes exactly where it left off without duplicating or skipping rows. That is why CREATE OR REPLACE PIPELINE is safe to re-run here.

• It can transform in flight. SET date = NOW() below is doing that, with no second pass over the table. Column subsets, expressions and computed columns work the same way.

• It keeps watching. FOREGROUND, used here, runs the load to completion and blocks — what you want in a notebook, so the next cell has data. BACKGROUND leaves the pipeline running, ingesting new files or new messages on a topic continuously with nobody triggering a job.

Watch it work in the next few cells: a million rows land in a few seconds, even on a Free Starter Workspace. You monitor it in SQL like anything else — SHOW PIPELINES, or the information_schema pipeline views for per-batch detail.

Data Preparation

Install UMAP, read the new readings out of sensor_data_stage into a DataFrame, and fill or drop null values so the embedding step has clean input.

UMAP (Uniform Manifold Approximation and Projection) is the dimensionality reduction library used in the next section.

Generating Vector Embeddings with UMAP

Take a slice of the new readings, and turn the three sensor values — vent, pluie, temp (wind, rain, temperature) — into a 15-dimensional vector each.

UMAP places readings that behave alike near each other in the new space, which is the property the scoring step depends on. The vectors are then L2-normalized, so dot_product() on a pair of them is their cosine similarity: 1.0 for identical, near 0 for unrelated.

Anomaly Detection with dot_product

For each new reading, one query does the work: dot_product() scores its embedding against every vector in sensor_data_with_vectors, the top 20 most similar rows are grouped by their anomaly label, and the label with the most votes wins. So the answer for a reading is one of vent, pluie or temp — wind, rain or temperature was the channel out of line — or none if the neighbours say the reading was ordinary. Readings with no similar history get a default value.

Note where the work happens — the scan, the sort and the cut to 20 all run in the database, so what crosses the network is a label, not a table of candidates.

The labelled rows are then cast to the table's column types and appended to sensor_data_with_vectors, which makes them part of the reference set for the next reading.

Dashboard: Anomalies over Time

Read the labelled rows back (skipping none), group them by day and anomaly type, and plot them with Plotly — first every city together, then one chart per city.

The stored labels are French, so they are translated to wind, rain and temperature for the chart legend. Each colour is therefore one sensor channel, and the height of a band is how often that channel was the one that looked wrong.

Real-Time Anomaly Detection Dashboard

display_table_contents() re-queries sensor_data_with_vectors every 10 seconds and redraws the 50 most recent readings as a rich table, colour-coded by anomaly type — the same view an operator would keep open.

Conclusion

What you built

A closed loop that labels new sensor readings by comparing them to readings that have already been judged.

Two pipelines loaded from S3: one filled sensor_data_with_vectors with historical readings that already carry an anomaly label and a vector embedding, the other filled sensor_data_stage with fresh readings stamped date = NOW(). For each new reading, dot_product() scored its embedding against every stored vector, the twenty nearest neighbours voted, and the winning label was written back into the history table. The dashboard then read that table live, so a row you labelled a moment ago is part of the reference set for the row after it.

There is no trained model here, and that is the point. Nothing was fitted, no thresholds were tuned, and no retraining step exists. "Anomalous" is defined entirely by what similar readings were called before, which means the definition improves as the labelled history grows rather than going stale between training runs.

Why it runs where the data lives

The scoring query never leaves the database. Each new reading is compared against the whole history, sorted, and cut to twenty rows before anything crosses the network, so what comes back is one label rather than a table of candidates. A few reasons that stays quick as the history grows:

• Queries are compiled to machine code. The per-row work in a dot_product scan is a tight machine-code loop, not an interpreter walking an expression tree once per row, and the compiled plan is cached for reuse.

• Vector math uses SIMD instructions. A CPU can multiply and accumulate several floats in a single instruction, which is exactly the shape of a similarity score.

• Partitions scan in parallel. The history is spread across partitions that each score their own slice and return only their best candidates to be merged. More compute means more throughput with no change to the query.

• Ingest and analytics hit the same tables. The pipelines write while the dashboard reads. There is no export step, no second system to keep in sync, and no window where the newest readings are invisible to scoring.

Where to take it next

• Swap S3 for a real stream. LOAD DATA KAFKA replaces LOAD DATA S3 and the rest of the notebook is unchanged, which is the lower panel of the architecture diagram.

• Add a vector index. At this size scoring every row is cheap. As the history grows into the millions, an index lets SingleStore approximate the nearest neighbours instead of scanning all of them, trading a measurable amount of recall for a much shorter scan.

• Use the labels, not just the vote. The twenty neighbours carry a similarity score. A weak top score is a reading unlike anything seen before, which is arguably the more interesting alert.

• Store the vectors as VECTOR. These tables keep embeddings in a TEXT column and unpack them with JSON_ARRAY_PACK at query time. A native VECTOR column skips that conversion and is what a new table should use.

Cleanup

The cell above runs until you stop it, so interrupt the kernel before continuing.

On a Standard or Premium Workspace the cell below drops iot_sensor_db, which removes both tables and both pipelines with it. A Free Starter Workspace cannot drop its own database, so there the same cell drops the two pipelines and the two tables instead — worth doing, because the historical load is around a million rows and sensor_data_with_vectors grows every time you re-run the labelling step. To release a Free Starter Workspace entirely, terminate it from the UI.