What this skill covers and why it matters
ML Training and Batch Pipelines are the backbone of production machine learning: sourcing data, generating features, training models, validating outputs, and writing artifacts on a reliable schedule. For MLOps Engineers, mastering this means you can ship models predictably, re-run history (backfills), meet SLAs, control cost, and keep data/model lineage auditable.
- Reliability: Idempotent, restartable, observable jobs.
- Reproducibility: Parameterized runs tied to data and code versions.
- Scalability: Distributed training and feature computation.
- Cost control: Smart use of resources, caching, and early stopping.
Who this is for
- MLOps Engineers automating model training and batch feature pipelines.
- Data/ML engineers moving from notebooks to production pipelines.
- Practitioners who need backfills, SLAs, retries, and alerts in real systems.
Prerequisites
- Solid Python and basic shell scripting.
- Familiarity with pandas or Spark for data processing.
- Basic ML training workflow (train/validate/save model).
- Comfort with containers and object storage paths (e.g., s3://, gs://, or equivalent).
Learning path (roadmap)
1) Pipeline templates & parameters
Design a single DAG that accepts run_date, data_version, and model_version. Use parameterized paths for inputs/outputs.
2) Ingestion & validation
Build robust readers, schema checks, and freshness controls before any feature or training step.
3) Feature generation
Create deterministic feature jobs. Version feature code and store artifacts with checksums or timestamps.
4) Training & evaluation
Train with reproducible seeds, log metrics, save models and training metadata. Add early stopping and resource limits.
5) Backfills, retries, alerts
Implement idempotent writes, retry policies, and alerting. Support backfill ranges safely.
6) Scheduling, SLA, and cost
Schedule by dependency, track SLAs, optimize costs with caching, sampling, and autoscaling.
Worked examples
Example 1 — Parameterized daily training pipeline (idempotent)
# Pseudocode: daily run parameterized by run_date
import os, json, datetime as dt
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import roc_auc_score
RUN_DATE = os.environ.get("RUN_DATE") # e.g., 2025-03-15
assert RUN_DATE, "RUN_DATE is required (YYYY-MM-DD)"
# Paths parameterized by run date (idempotency)
RAW_PATH = f"/data/raw/dt={RUN_DATE}/events.parquet"
FEAT_PATH = f"/data/features/dt={RUN_DATE}/features.parquet"
MODEL_PATH = f"/models/date={RUN_DATE}/model.pkl" # no overwrites across runs
METRICS_PATH = f"/metrics/date={RUN_DATE}/metrics.json"
# Ingestion
pdf = pd.read_parquet(RAW_PATH)
assert len(pdf) > 0, "No data for run_date"
# Minimal validation
assert all(c in pdf.columns for c in ["label","f1","f2","f3"])
# Feature step
pdf["f_ratio"] = (pdf["f1"] + 1) / (pdf["f2"] + 1)
pdf.to_parquet(FEAT_PATH)
# Train
X = pdf[["f1","f2","f3","f_ratio"]]
y = pdf["label"]
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
# Eval with simple holdout
p = model.predict_proba(X)[:,1]
auc = roc_auc_score(y, p)
# Persist artifacts
import joblib, pathlib
pathlib.Path(os.path.dirname(MODEL_PATH)).mkdir(parents=True, exist_ok=True)
pathlib.Path(os.path.dirname(METRICS_PATH)).mkdir(parents=True, exist_ok=True)
joblib.dump(model, MODEL_PATH)
json.dump({"run_date": RUN_DATE, "auc": auc}, open(METRICS_PATH, "w"))
print("DONE", RUN_DATE, "AUC=", round(auc, 4))
Key points: run_date parameterization, deterministic outputs, basic validation, and separate paths for metrics and model.
Example 2 — Data validation gate
# Quick validation before feature/training
import pandas as pd
def validate(df: pd.DataFrame):
# Schema
required = {"label": "int64", "f1": "float64", "f2": "float64", "f3": "float64"}
for col, dtype in required.items():
assert col in df.columns, f"Missing column {col}"
assert str(df[col].dtype) == dtype, f"Wrong dtype for {col}"
# Ranges & missing
assert df[["f1","f2","f3"]].isna().mean().max() < 0.01, "Too many nulls"
assert df["label"].between(0,1).all(), "Labels must be 0/1"
# Drift sentinel (simple mean check)
historical_mean_f1 = 0.5 # from prior runs metadata
assert abs(df["f1"].mean() - historical_mean_f1) < 0.2, "f1 drift too high"
return True
Fail fast on schema or drift. Emit clear error messages so retries/alerts are meaningful.
Example 3 — Distributed training basics (conceptual pattern)
# Pseudocode outline for distributed training (e.g., PyTorch DDP)
# Orchestrator sets WORKERS=N, RUN_DATE, and data shard paths.
# 1) Shard data by consistent hash or date split
# 2) Launch N workers with the same code revision and seed
# 3) Aggregate metrics/checkpoints centrally
# Tuning tips:
# - Increase global batch size ~ linearly with workers; scale learning rate.
# - Enable gradient accumulation if memory bound.
# - Implement early stopping and max wall-clock time.
# - Save checkpoints to /checkpoints/date=RUN_DATE/step=K/ to resume.
Goal: scale with more workers while keeping training reproducible and resumable.
Example 4 — Safe backfill for a date range
from datetime import date, timedelta
def daterange(start, end):
d = start
while d <= end:
yield d
d += timedelta(days=1)
# Backfill Mar 1 to Mar 7 (inclusive)
start, end = date(2025,3,1), date(2025,3,7)
for d in daterange(start, end):
# Submit the same pipeline with RUN_DATE=str(d)
# Each run writes to date-partitioned outputs. Idempotent re-runs overwrite atomically.
submit_pipeline(run_date=str(d))
Backfills reuse the exact same DAG with different parameters. Ensure outputs are partitioned by run_date to avoid collisions.
Example 5 — Failure handling, retries, and atomic writes
import tempfile, os, shutil
TARGET = "/models/date=2025-03-15/model.pkl"
TMP_DIR = tempfile.mkdtemp()
TMP_FILE = os.path.join(TMP_DIR, "model.pkl")
try:
# Train and write to temp
train_model_to(TMP_FILE)
# Atomic move (best-effort on POSIX or via storage rename)
os.makedirs(os.path.dirname(TARGET), exist_ok=True)
shutil.move(TMP_FILE, TARGET)
except TransientError as e:
# Let the orchestrator retry with exponential backoff
raise
finally:
shutil.rmtree(TMP_DIR, ignore_errors=True)
Write-then-rename prevents partial files. Combine with a retry policy and clear error types to distinguish transient vs permanent failures.
Drills and exercises
- Make your current training script accept run_date, data_version, and output_dir as parameters.
- Add schema validation that fails fast with clear messages.
- Ensure all outputs are partitioned or versioned so re-runs do not overwrite incorrectly.
- Implement early stopping and persist best checkpoint with its metrics.
- Add a retry policy (at least 3 attempts, exponential backoff) to ingestion and model upload steps.
- Create a backfill script for a 14-day window and prove it’s idempotent by re-running a subset.
- Track simple drift stats (mean, std) for two key features across the last 7 runs.
- Introduce a cost guardrail: cap max epochs or training wall time by environment variable.
Common mistakes and debugging tips
- Non-idempotent outputs: Using a fixed path like model-latest.pkl causes overwrites. Fix by partitioning with run_date or model_version.
- Hidden time dependence: Using now() inside code breaks reproducibility. Pass run_date as a parameter.
- Skipping validation: Train runs succeed on corrupt data, then fail in serving. Add schema and sanity checks early.
- Unbounded retries: Infinite retries hide real issues and burn cost. Limit attempts and alert when thresholds are hit.
- No atomic writes: Partial files cause downstream errors. Write to temp, then move/commit.
- Inconsistent feature logic: Train and serve drift apart. Centralize feature definitions or version them together.
- Distributed training with same hyperparams: Larger world size needs adjusted learning rate/batch size. Add scaling rules.
Mini project: Daily churn model with safe backfills
Build a daily pipeline that ingests user events, validates schema, generates features, trains a classifier, evaluates metrics, and writes artifacts partitioned by run_date.
- Parameters: run_date, data_version, model_version, max_epochs, budget_minutes.
- Steps: ingest → validate → feature → train → eval → publish (model, metrics, lineage).
- Backfill: a CLI that runs the same DAG for a date range with concurrency of 3.
- Reliability: retries on network steps; alerts when validation fails or AUC drops >= 5% vs 7-day average.
- Cost: add early stopping, cap training minutes, and optional data sampling (e.g., 30%) in dev.
Acceptance checklist
- All outputs stored at /models/date=YYYY-MM-DD/ and /metrics/date=YYYY-MM-DD/.
- Validation blocks bad schema or drift beyond thresholds.
- Rerunning the same date produces identical artifacts (hash/equality check).
- Backfill over 14 days completes with retries and logs a summary.
- Alerts fire when AUC decreases ≥ 0.05 vs rolling mean.
Subskills
- Pipeline Templates And Parameterization — Build reusable DAGs with run parameters for date, data version, and outputs.
- Data Ingestion And Validation — Robust readers, schema checks, and freshness/drift gates.
- Feature Generation Steps — Deterministic, versioned feature pipelines aligned with training and serving.
- Distributed Training Basics — Scale out training safely with checkpoints and reproducibility.
- Backfills And Idempotency — Re-run history safely using partitioned outputs and stable parameters.
- Failure Handling Retries Alerts — Classify errors, retry transient ones, and alert on meaningful conditions.
- Scheduling And SLA Management — Reliable schedules, dependencies, and SLA tracking.
- Cost Aware Training Runs — Control spend with sampling, early stopping, and resource caps.
Next steps
- Automate metadata logging: commit git SHA, data snapshots, and hyperparams in every run.
- Introduce a simple feature registry so training and serving share definitions.
- Add a weekly retraining job with longer compute windows and thorough validation.
- Prepare for model serving and monitoring (latency, drift, and performance SLAs).