Data Engineering for Machine Learning and Analytics

# From Raw Data to ML‑Ready Features

The moment a sensor in a wind farm spits out a 12‑byte protobuf packet, the downstream system must decide whether that datum will ever see a model. In practice the decision happens in milliseconds: a lightweight validator checks schema compliance, a schema registry flags version mismatches, and a Kafka topic routes the record to a stream processor that enriches it with location metadata. The enrichment step is where the raw signal becomes useful; without it the model would only see a flat vector of numbers that lack context.

# Understanding the Data Journey

Every ML project starts with a data contract. In the platform we built after Chapter 6, each contract lives in a Confluent Schema Registry as an Avro definition. The registry guarantees that producers and consumers speak the same language, and it also provides compatibility checks that prevent accidental breaking changes. For example, a temperature sensor team added a new field battery_voltage with a default of null. Because the registry was set to BACKWARD_TRANSITIVE, older consumers continued to run without modification, while the new feature could be consumed by downstream pipelines as soon as they upgraded.

The next step is persistence. Chapter 3 taught us that raw data should be stored in an immutable lake, typically an S3‑compatible bucket with a Delta Lake table on top. Delta Lake gives us ACID guarantees and time travel, which means a data scientist can rewind to the exact snapshot that produced a model version. In practice we create a daily partition on event_date and a secondary partition on device_id. The partition layout looks like:

s3://analytics-data/
  └─ raw/
      └─ temperature/
          ├─ event_date=2025-07-01/
          │   └─ device_id=12345/
          │       └─ part-00001.parquet
          └─ event_date=2025-07-02/
              └─ device_id=67890/
                  └─ part-00002.parquet

This layout supports predicate pushdown: a query that filters on a specific device and date touches only a handful of files, keeping latency low even as the table grows to billions of rows.

# Feature Engineering at Scale

Feature engineering is where domain knowledge meets distributed compute. In our platform we use a combination of Spark Structured Streaming and Flink SQL to calculate rolling aggregates, categorical encodings, and statistical summaries. A typical rolling average for a turbine’s vibration metric looks like this:

$${\mu }_t=\frac{1}{W}\sum _{i=0}^{W-1}{x}_{t-i}$$ ${\mu }_t=\frac{1}{W}\sum _{i=0}^{W-1}{x}_{t-i}$" title="Ask AI about this paragraph">

where (W) is the window size in seconds. In Spark we express the same logic with a watermark and a tumbling window:

# spark_feature_engineering.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import avg, col, window

spark = SparkSession.builder.appName("FeatureEngineering").getOrCreate()

# Read from the raw Delta table
raw = spark.readStream.format("delta").load("s3://analytics-data/raw/temperature")

# Compute a 5‑minute rolling average per turbine
features = (
    raw
    .withWatermark("event_timestamp", "10 minutes")
    .groupBy(
        col("device_id"),
        window(col("event_timestamp"), "5 minutes")
    )
    .agg(avg(col("vibration")).alias("vibration_avg"))
)

# Write feature stream to a new Delta table
query = (
    features.writeStream
    .format("delta")
    .option("checkpointLocation", "s3://analytics-data/checkpoints/vibration_avg")
    .outputMode("append")
    .trigger(processingTime="1 minute")
    .start("s3://analytics-data/features/vibration_avg")
)

query.awaitTermination()

The code uses the spark.readStream API introduced in Spark 3.5, which is the current stable release as of 2025. The checkpoint location lives in the same bucket, ensuring exactly‑once semantics even if the job restarts. Notice the use of withWatermark; without it the state would grow without bound, eventually exhausting memory.

When the feature store is ready, data scientists pull the latest snapshot via a simple SQL query:

SELECT *
FROM delta.`s3://analytics-data/features/vibration_avg`
WHERE device_id = '12345'
  AND event_timestamp BETWEEN '2025-07-01' AND '2025-07-02';

Because the feature table is partitioned by device_id and event_date, this query returns in seconds, not minutes.

# Model Training Pipelines

Feature readiness is only half the story. Training must be reproducible, versioned, and scalable across GPUs or TPUs. The platform we described in Chapter 6 already runs workloads on a Kubernetes cluster with KEDA‑driven autoscaling. For model training we add a custom resource definition (CRD) called MLJob that abstracts the underlying engine - whether it is PyTorch, TensorFlow, or XGBoost.

# Orchestrating Distributed Training

A typical MLJob manifest for a PyTorch distributed run looks like this:

# mljob_pytorch.yaml
apiVersion: ml.k8s.io/v1
kind: MLJob
metadata:
  name: turbine-failure-pytorch
spec:
  framework: pytorch
  version: "2.2"
  replicas: 4               # 4‑node data‑parallel training
  resources:
    limits:
      nvidia.com/gpu: 1
  command: ["python", "train.py"]
  args:
    - "--data-path"
    - "s3://analytics-data/features/"
    - "--epochs"
    - "30"
    - "--batch-size"
    - "256"
  env:
    - name: WANDB_API_KEY
      valueFrom:
        secretKeyRef:
          name: wandb-secret
          key: api-key
  # Enable KEDA scaling based on queue length in a training job queue
  scaling:
    minReplicas: 1
    maxReplicas: 8
    triggers:
      - type: kafka
        metadata:
          bootstrapServers: kafka-broker:9092
          topic: training-jobs
          lagThreshold: "10"

The manifest targets Kubernetes 1.30 LTS, the version we lock for production. The scaling block uses KEDA 2.10+, which watches a Kafka topic for pending training jobs and scales the MLJob up or down automatically. This pattern eliminates the need for manual capacity planning; the cluster expands only when the queue length exceeds ten messages.

Inside train.py we rely on PyTorch Lightning to handle the distributed backend. The script logs metrics to Weights & Biases (W&B) using a short‑lived token fetched from a Kubernetes secret, satisfying the security requirements outlined in Chapter 5. Because the token expires after 12 hours, even a compromised pod cannot reuse the credential indefinitely.

# Managing Experiment Metadata

Experiment tracking is a cross‑cutting concern. In our platform we store every run’s hyperparameters, data version, and model artifact in a PostgreSQL instance that lives behind a private VPC. The schema is simple:

CREATE TABLE experiments (
    id UUID PRIMARY KEY,
    name TEXT NOT NULL,
    start_ts TIMESTAMPTZ NOT NULL,
    end_ts TIMESTAMPTZ,
    params JSONB NOT NULL,
    metrics JSONB,
    model_uri TEXT,
    git_commit TEXT NOT NULL
);

When a training job finishes, a sidecar container pushes a row into this table. The sidecar uses psycopg2-binary version 2.9.9, which is the latest stable as of early 2025. The sidecar also verifies that the model_uri points to an object in an S3 bucket protected by bucket‑level policies that enforce TLS 1.3 encryption in transit and KMS‑managed keys at rest. This chain of trust mirrors the defense‑in‑depth approach from Chapter 5.

# Serving and Monitoring

A model is only useful when it can answer queries in production. Serving introduces new challenges: latency, version management, and drift detection. Our platform adopts KServe (formerly KFServing) for model deployment, which gives us canary rollouts, autoscaling, and out‑of‑the‑box observability.

# Deploying Models as Services

A KServe InferenceService for a binary classifier that predicts turbine failure looks like this:

# inference_service.yaml
apiVersion: "serving.kserve.io/v1beta1"
kind: InferenceService
metadata:
  name: turbine-failure
  annotations:
    # Enable canary rollout with 10% traffic to new version
    autoscaling.knative.dev/target: "100"
spec:
  predictor:
    canaryTrafficPercent: 10
    model:
      modelFormat:
        name: torchscript
      storageUri: "s3://model-registry/turbine-failure/v1.3/"
      resources:
        limits:
          cpu: "2"
          memory: "4Gi"
          nvidia.com/g