How to Use SQL for ML Feature Engineering on Time-Series Sensor Data

You have 500,000 sensor readings across 500 devices. You need per-device rolling averages, standard deviations, and fault rates as input features for a fault classifier. The standard approach: load the dataset into pandas, group by device, compute rolling windows. At the feature engineering step, memory usage peaks and the model trains. Then new readings arrive and you need to retrain. That means exporting the dataset again first.

This is not a Python problem. It is a feature engineering architecture problem.

Why rolling features carry more signal than raw readings

A fault classifier trained only on raw sensor values performs poorly in production. The reason is structural: a device reading metric_value = 85 means something different depending on whether it has held steady at 85 for the past 20 readings or climbed from 30 in the past six hours. The raw number cannot tell you which is happening. Rolling statistics can.

The features that consistently outrank raw readings in importance scores for industrial sensor models:

When trained on a 500-device, 500,000-reading dataset, fault_rate_5 is the strongest single predictor. If 3 of the last 5 readings for a device were warning or critical, the device is on a fault trajectory with high confidence. Rolling mean and standard deviation rank second and fourth. The raw metric_value at reading time ranks considerably lower.

The point: context over time is the signal. Point-in-time values are noise.

The pandas approach and where it breaks

Here is what building these features looks like in Python against a JSON dataset:

import pandas as pd, json

# Load all 500,000 rows — peaks at ~4 GB RAM (internal benchmark)
with open('dataset/iot_demo_dataset.json', 'r') as f:
    df = pd.DataFrame([json.loads(line) for line in f])

df = df.sort_values(['device_id', 'timestamp'])

# Compute rolling features per device
df = df.groupby('device_id', group_keys=False).apply(lambda g: g.assign(
    metric_mean_5  = g['metric_value'].rolling(5,  min_periods=1).mean(),
    metric_std_5   = g['metric_value'].rolling(5,  min_periods=1).std(),
    metric_mean_10 = g['metric_value'].rolling(10, min_periods=1).mean(),
    metric_std_10  = g['metric_value'].rolling(10, min_periods=1).std(),
    fault_rate_5   = (g['status'].isin(['warning','critical'])
                      .astype(int).rolling(5,  min_periods=1).mean()),
    fault_rate_10  = (g['status'].isin(['warning','critical'])
                      .astype(int).rolling(10, min_periods=1).mean()),
    metric_delta   = g['metric_value'].diff(),
))

This works for a one-time run on a static file. At production scale, the approach breaks in three specific ways: memory, re-export before retraining, and feature freshness.

The re-export requirement is the highest-risk failure point. A scheduled retraining job that forgets to re-export first trains the model on data that is days or weeks stale. The model's performance degrades silently without surfacing a pipeline error.

Pre-aggregating in CrateDB SQL

CrateDB's DATE_BIN function bins timestamps into fixed-width time windows. Combined with STDDEV, AVG, and the SQL FILTER clause for conditional aggregation, it pre-computes the same rolling statistics in the database, before any data crosses the network.

-- Feature engineering: per-device hourly windows with fault rate
-- DATE_BIN groups each reading into its 1-hour window
-- COUNT(*) FILTER computes fault_rate without a separate join or Python transform
-- Returns one feature row per device per hour: compact, ready for training

SELECT
    device_id,
    device_type,
    plant_id,
    DATE_BIN('1 hour'::INTERVAL, "timestamp", TIMESTAMP '2025-09-01') AS window_start,
    AVG(metric_value) AS metric_mean,
    STDDEV(metric_value) AS metric_std,
    AVG(quality_score) AS quality_mean,
    COUNT(*) FILTER (WHERE status IN ('warning', 'critical'))
        * 1.0 / NULLIF(COUNT(*), 0) AS fault_rate,
    MAX(CASE WHEN status IN ('warning', 'critical') THEN 1 ELSE 0 END) AS had_fault
FROM iot_data
WHERE "timestamp" >= TIMESTAMP '2025-09-01'
GROUP BY device_id, device_type, plant_id, window_start
ORDER BY device_id, window_start;

Three SQL constructs do the work that the pandas code does in 40+ lines:

DATE_BIN('1 hour'::INTERVAL, "timestamp", origin): bins each reading into its one-hour window. The origin anchor aligns bins consistently across devices and across retraining runs. Change '1 hour' to '15 minutes' or '1 day' to adjust window granularity without changing any other part of the query.

STDDEV(metric_value): standard deviation per window per device. Volatile devices show rising STDDEV in the windows before a fault. This maps directly to metric_std_5 and metric_std_10 in the pandas approach, expressed over time windows rather than rolling row counts.

COUNT(*) FILTER (WHERE status IN ('warning', 'critical')): counts only the readings in a fault state within that window, then divides by total readings to produce a fault rate. This is the window-based equivalent of fault_rate_5. The FILTER clause is standard SQL and avoids a separate CASE WHEN subquery or a Python-side computation.

The NULLIF(COUNT(*), 0) guard prevents division by zero on windows that contain only one reading.

The query returns one row per device per hour. Each row is a feature vector. No rolling transform needed in Python. had_fault is the training target: 1 if any reading in that window was in warning or critical status.

From SQL feature table to trained model

pip install crate sqlalchemy pandas xgboost scikit-learn
import pandas as pd
from sqlalchemy import create_engine
from xgboost import XGBClassifier
from sklearn.model_selection import train_test_split

# Point at your CrateDB instance
engine = create_engine("crate://localhost:4200")
# CrateDB Cloud: "crate://admin:@.cratedb.net:4200"

# Pull the feature table: compact, pre-aggregated
df = pd.read_sql(sql, engine)  # sql = the query above

# Train
X = df[['metric_mean', 'metric_std', 'quality_mean', 'fault_rate']]
y = df['had_fault']

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)
model = XGBClassifier(scale_pos_weight=(y_train == 0).sum() / (y_train == 1).sum())
model.fit(X_train, y_train)

The DataFrame arrives compact. At 500 devices with hourly readings over 90 days, the feature table contains 500 × 24 × 90 = 1,080,000 rows, one per device-hour window. That is the same information derived from 500,000 raw readings, returned as a pre-aggregated DataFrame that fits in memory without issue.

Scheduled retraining without the export step

Scope the SQL query by time window and the pattern becomes a scheduled retraining pipeline:

-- Same feature query scoped to the last 90 days
-- Run on a cron schedule: no export required

SELECT
    device_id,
    device_type,
    plant_id,
    DATE_BIN('1 hour'::INTERVAL, "timestamp", TIMESTAMP '2025-09-01') AS window_start,
    AVG(metric_value) AS metric_mean,
    STDDEV(metric_value) AS metric_std,
    AVG(quality_score) AS quality_mean,
    COUNT(*) FILTER (WHERE status IN ('warning', 'critical'))
        * 1.0 / NULLIF(COUNT(*), 0) AS fault_rate,
    MAX(CASE WHEN status IN ('warning', 'critical') THEN 1 ELSE 0 END) AS had_fault
FROM iot_data
WHERE "timestamp" >= NOW() - INTERVAL '90' DAY
GROUP BY device_id, device_type, plant_id, window_start
ORDER BY device_id, window_start;

A weekly retraining job runs this query, trains the model on the fresh feature windows, and saves the updated model file. No stale snapshot. No export job to schedule separately. The model reflects the fleet's current fault patterns as of training time, not the fleet's behavior as of the last time someone ran an export.

CrateDB distributes the aggregation across shards before returning results. The data ingestion architecture that feeds iot_data in real time (Telegraf, Kafka, MQTT) is covered in the IoT Analytics Architecture Guide. For multi-plant deployments, the cross-plant query pattern from the Cross-Plant Visibility guide extends to training a per-plant or plant-group model using the same DATE_BIN + FILTER pattern with a WHERE plant_id = 'PLANT_007' filter.

Taking the trained model to production

Feature engineering in SQL removes the export step and the memory bottleneck from the training side. The inference side has a different problem: when a new reading arrives, the model needs the last 50 readings for that device as rolling-window context before it can score. That context lives in CrateDB, and because iot_data is CLUSTERED BY (device_id), fetching those 50 rows hits a single shard and returns in single-digit milliseconds.

The full inference architecture (FastAPI service, per-device scoring endpoint, fault_probability written back to CrateDB and JOIN-able against live sensor readings) is in Real-Time ML Inference on Live Sensor Data: Architecture with CrateDB and FastAPI.

For the anomaly detection model (Isolation Forest trained on healthy-fleet readings to score statistical deviation independent of fault labels), see Anomaly Detection on Industrial Sensor Data: Beyond Threshold Alerts.

The database architecture that supports both pipelines, including the fault_predictions table schema and the predictive maintenance SQL patterns that run continuously against live data, is in the Predictive Maintenance Database Architecture guide.

Feature engineering that lives in SQL rather than Python removes the export step, the memory bottleneck, and the freshness gap between sensor stream and training data. Run queries on live sensor data to see how DATE_BIN and conditional aggregates perform on a working dataset.