Data Workflows: NumPy, Pandas, SQL, and Data Quality

Data Reliability Is an Engineering Problem

Software engineers encounter well-structured inputs - JSON from an API, rows from a typed ORM, messages from a queue. ML engineers encounter raw data: CSVs with inconsistent encoding, databases with undocumented schemas, event logs with missing timestamps.

The challenge is not just cleaning this data. It is building a data pipeline that fails loudly when the data is bad, documents its assumptions, and produces model inputs that are reliable enough to trust the downstream model's behavior.

NumPy for ML Computations

NumPy arrays are the standard container for ML computations. Key patterns:

pythonCopy
import numpy as np

# Create arrays
X = np.random.randn(1000, 50).astype(np.float32)  # float32 for GPU compatibility
y = np.random.randint(0, 2, size=1000)

# Normalization (vectorized, no loops)
X_mean = X.mean(axis=0)      # Per-feature mean, shape (50,)
X_std = X.std(axis=0)        # Per-feature std, shape (50,)
X_norm = (X - X_mean) / (X_std + 1e-8)  # Epsilon avoids divide-by-zero

# Boolean indexing for filtering
positive_mask = y == 1
X_positive = X[positive_mask]   # Rows where y = 1

# Stacking arrays
X_train = np.vstack([X[:800], X_positive[:50]])    # Add more positives to train
y_train = np.concatenate([y[:800], y[positive_mask][:50]])

Float precision: Use float32 for model inputs (sufficient precision, half the memory of float64, GPU-native). Use float64 for statistical computations where precision matters.

Pandas: Real-World Data Manipulation

pythonCopy
import pandas as pd
from pathlib import Path

# Load with type inference assistance
df = pd.read_csv("data/raw/events.csv",
                 parse_dates=["event_timestamp"],
                 dtype={"user_id": str, "amount": float})

# Always inspect before anything else
print(df.shape)
print(df.dtypes)
print(df.describe())
print(df.isnull().sum().sort_values(ascending=False))
print(df["label"].value_counts(normalize=True))

Feature Engineering Patterns

pythonCopy
# 1. Aggregation: from event-level to entity-level
user_features = df.groupby("user_id").agg(
    n_events=("event_id", "count"),
    total_spend=("amount", "sum"),
    avg_spend=("amount", "mean"),
    recency_days=("event_timestamp", lambda x: (pd.Timestamp.now() - x.max()).days),
    first_event=("event_timestamp", "min"),
    last_event=("event_timestamp", "max"),
).reset_index()

# 2. Time features from datetime
df["hour_of_day"] = df["event_timestamp"].dt.hour
df["day_of_week"] = df["event_timestamp"].dt.dayofweek
df["is_weekend"] = (df["day_of_week"] >= 5).astype(int)

# 3. Cumulative (temporal) features - no leakage
df = df.sort_values("event_timestamp")
df["cumulative_spend"] = df.groupby("user_id")["amount"].cumsum()
df["event_rank"] = df.groupby("user_id").cumcount() + 1

# 4. Lag features
df["prev_amount"] = df.groupby("user_id")["amount"].shift(1)

Handling Missing Values

pythonCopy
# Step 1: Understand the pattern
null_summary = pd.DataFrame({
    "null_count": df.isnull().sum(),
    "null_pct": df.isnull().mean().round(4),
    "dtype": df.dtypes,
}).query("null_count > 0").sort_values("null_pct", ascending=False)
print(null_summary)

# Step 2: Add indicators before imputing
for col in ["income", "age"]:
    df[f"{col}_was_missing"] = df[col].isna().astype(int)

# Step 3: Impute (fit on train, transform all splits)
from sklearn.impute import SimpleImputer

imputer = SimpleImputer(strategy="median")
df[numeric_cols] = imputer.fit_transform(df[numeric_cols])
# Save the fitted imputer: joblib.dump(imputer, "models/imputer.pkl")

SQL for Feature Extraction

SQL is the right place to join, aggregate, and filter before loading data into Python:

sqlCopy
-- Training dataset with temporal leakage prevention
WITH
prediction_window AS (
    SELECT
        user_id,
        subscription_start    AS prediction_date,
        cancelled_within_30d  AS label
    FROM subscriptions
    WHERE subscription_start BETWEEN '2023-01-01' AND '2024-03-31'
),

prior_activity AS (
    SELECT
        pw.user_id,
        pw.prediction_date,
        COUNT(e.event_id)                                  AS event_count_prior,
        COALESCE(SUM(e.amount), 0)                        AS total_spend_prior,
        MAX(e.event_date)                                  AS last_event_prior,
        EXTRACT(DAY FROM pw.prediction_date - MAX(e.event_date)) AS days_since_last
    FROM prediction_window pw
    LEFT JOIN events e
        ON pw.user_id = e.user_id
        AND e.event_date < pw.prediction_date         -- Only prior data
    GROUP BY pw.user_id, pw.prediction_date
)

SELECT pw.user_id, pw.prediction_date, pw.label,
       pa.event_count_prior, pa.total_spend_prior,
       pa.days_since_last
FROM prediction_window pw
LEFT JOIN prior_activity pa USING (user_id, prediction_date);

The AND e.event_date < pw.prediction_date clause is the temporal leakage guard. Without it, features would include post-label data, producing a model that performs well offline and poorly in production.

Data Quality Validation

Data quality checks should be automated and run before any modeling:

pythonCopy
from dataclasses import dataclass
from typing import Any
import logging

logger = logging.getLogger(__name__)

@dataclass
class ColumnRule:
    col: str
    nullable: bool = False
    min_val: float | None = None
    max_val: float | None = None
    allowed_vals: list[Any] | None = None
    max_null_frac: float = 0.0

def validate(df: pd.DataFrame, rules: list[ColumnRule]) -> list[str]:
    errors = []
    for r in rules:
        if r.col not in df.columns:
            errors.append(f"MISSING COLUMN: {r.col}")
            continue
        null_frac = df[r.col].isna().mean()
        if not r.nullable and null_frac > r.max_null_frac:
            errors.append(f"{r.col}: {null_frac:.1%} nulls (max allowed: {r.max_null_frac:.1%})")
        if r.min_val is not None and df[r.col].min() < r.min_val:
            errors.append(f"{r.col}: min={df[r.col].min()} below {r.min_val}")
        if r.max_val is not None and df[r.col].max() > r.max_val:
            errors.append(f"{r.col}: max={df[r.col].max()} above {r.max_val}")
        if r.allowed_vals and df[r.col].dropna().isin(r.allowed_vals).mean() < 1.0:
            bad = df[r.col].dropna()[~df[r.col].dropna().isin(r.allowed_vals)].unique()[:5]
            errors.append(f"{r.col}: unexpected values {bad}")
    return errors

RULES = [
    ColumnRule("user_id", nullable=False),
    ColumnRule("event_count_prior", nullable=False, min_val=0),
    ColumnRule("total_spend_prior", nullable=False, min_val=0),
    ColumnRule("label", nullable=False, allowed_vals=[0, 1]),
]

errors = validate(training_df, RULES)
if errors:
    msg = "Data validation failed:\n" + "\n".join(errors)
    logger.error(msg)
    raise ValueError(msg)
logger.info("Data validation passed: %d rows", len(training_df))

Run this before fitting any model. Silent data quality problems are among the hardest ML bugs to diagnose because they cause model degradation rather than exceptions.

Leakage: The Data Preparation Bug With the Highest Cost

Leakage produces models that look correct in evaluation and fail in production. It is the most expensive category of data preparation bug.

Three forms:

1. Pipeline leakage - fitting a preprocessor on data that includes the validation set:

pythonCopy
# WRONG
scaler = StandardScaler().fit(X)          # Sees validation data
X_train_s = scaler.transform(X_train)

# CORRECT
scaler = StandardScaler().fit(X_train)    # Only training data
X_train_s = scaler.transform(X_train)
X_val_s = scaler.transform(X_val)        # Transform with training statistics

2. Temporal leakage - features computed using events that happen after the prediction time (shown in the SQL example above).

3. Target leakage - a feature derived from or caused by the target. Example: using support_tickets_filed to predict is_churned - tickets are filed as a result of churn behavior.

The diagnostic question for every feature: At the moment I would use this model in production, would I actually have this feature value?

Where to Go Next

With Modules 1–4 complete, you have the mental models, Python tooling, math intuition, and data skills to work effectively in ML systems. Module 5 (Supervised Learning End to End) begins the core ML engineering curriculum: end-to-end supervised workflows with proper evaluation, baselines, and iteration discipline.