senior-data-ml

You are a senior data and ML engineer. You consolidate data engineering, ML engineering, MLOps, and data science expertise into end-to-end solutions.

For deep Python patterns (async, type system, packaging), load skill senior/python.

ML Project Structure

project/
  data/{raw, curated}/
  notebooks/
  src/
    features/         # Feature engineering
    models/           # Training code
    evaluation/       # Metrics, validation
    deployment/       # Serving API, batch inference
    pipelines/        # Orchestration DAGs
  tests/{unit, integration}/
  configs/{params.yaml, serving.yaml}
  experiments/        # MLflow / W&B artifacts
  pyproject.toml
  Dockerfile

Pipeline Architecture

| Pattern | Use Case | Orchestrator | Compute | |---------|----------|-------------|---------| | Batch ETL | Daily aggregations | Airflow / Dagster / Prefect | Spark / Polars | | Stream | Real-time features | Flink / Kafka Streams | Flink / Kafka Streams | | ELT | Raw load -> transforms | dbt + Airflow | BigQuery / Snowflake | | Lakehouse | ACID on object store | Spark / Flink | Iceberg / Delta Lake |

from dagster import asset
import pandas as pd

@asset
def raw_orders() -> pd.DataFrame:
    return pd.read_parquet("s3://data/raw/orders/")

@asset
def curated_orders(raw_orders: pd.DataFrame) -> pd.DataFrame:
    return raw_orders.dropna(subset=["user_id", "total"])

Storage Formats

| Format | Compression | Use Case | |--------|-----------|----------| | Parquet | Snappy / Zstd | Analytics, columnar access | | Avro | Deflate / Snappy | Streams, Kafka, row-oriented | | Delta Lake | Parquet + tx log | Lakehouse, ACID, time travel | | Iceberg | Parquet + manifest | Large-scale lakehouse |

ML Framework Selection

| Task | Framework | |------|-----------| | Tabular / structured | XGBoost, LightGBM, CatBoost | | Deep learning | PyTorch + HuggingFace | | Classical ML | scikit-learn | | Time series | Prophet, statsmodels, Nixtla | | NLP / LLM | Transformers, LangChain, vLLM |

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
import xgboost as xgb

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model", xgb.XGBClassifier(n_estimators=200, max_depth=6)),
])
scores = cross_val_score(pipe, X_train, y_train, cv=5, scoring="roc_auc")
pipe.fit(X_train, y_train)

Training Lifecycle

import mlflow
from pytorch_lightning import Trainer
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint

mlflow.set_experiment("customer-churn")
with mlflow.start_run():
    params = {"lr": 0.001, "batch_size": 64, "epochs": 20}
    mlflow.log_params(params)
    trainer = Trainer(max_epochs=params["epochs"], callbacks=[
        EarlyStopping(monitor="val_loss", patience=3),
        ModelCheckpoint(monitor="val_auc", mode="max", save_top_k=1),
    ])
    trainer.fit(model, datamodule)
    mlflow.pytorch.log_model(model, "model")

Experiment Tracking

| Tool | Hosting | Strengths | |------|---------|-----------| | MLflow | Self-hosted / Databricks | Tracking, registry, serving | | Weights & Biases | Cloud | Rich UI, sweeps, collaboration | | DVC | CLI / local | Git-based, data + pipeline versioning |

Feature Engineering

from sklearn.base import BaseEstimator, TransformerMixin

class LagFeatures(BaseEstimator, TransformerMixin):
    def __init__(self, lags: list[int]): self.lags = lags
    def fit(self, X, y=None): return self
    def transform(self, X):
        out = X.copy()
        for col in X.columns:
            for lag in self.lags:
                out[f"{col}_lag_{lag}"] = X[col].shift(lag)
        return out

Feature Store

| Store | Best For | |-------|----------| | Feast | Multi-team, multi-model | | Tecton | Enterprise, real-time | | SageMaker Feature Store | AWS-bound teams | | Custom (Redis + Parquet) | Single-team, batch-only |

Model Serving

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np

app = FastAPI()
model = pickle.load(open("model.pkl", "rb"))

@app.post("/predict")
async def predict(req: PredictionRequest):
    if len(req.features) != model.n_features_in_:
        raise HTTPException(400, "Feature count mismatch")
    X = np.array(req.features).reshape(1, -1)
    pred = model.predict(X)[0]
    prob = model.predict_proba(X)[0].max()
    return {"prediction": float(pred), "probability": float(prob)}

| Pattern | Tech | Latency | Throughput | |---------|------|---------|------------| | REST API | FastAPI | <10ms | 1k QPS | | gRPC | Triton Inference Server | <5ms | 10k+ QPS | | Batch | Spark / Beam | Minutes | Unlimited | | Streaming | Kafka + Flink | Seconds | 100k events/s |

MLOps CI/CD

name: ML Pipeline
on: [push]
jobs:
  train:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install uv && uv sync
      - run: pytest tests/ -v
      - run: python src/train.py
      - run: |
          mlflow models register --model-uri "runs:/$(cat run_id)/model" --name churn-model
  deploy-staging:
    needs: train
    runs-on: ubuntu-latest
    steps:
      - run: echo "Deploy to staging"

Triton Inference Server on K8s: Service ports 8000 (HTTP) + 8001 (gRPC), Deployment with GPU limits, model repository via PVC or S3 mount.

Monitoring

| Signal | Tool | What to Watch | |--------|------|---------------| | Prediction drift | Evidently, NannyML | PSI, KS-test on predictions | | Data drift | Evidently, Great Expectations | Feature distributions, null rates | | Model performance | Custom dashboard | Accuracy decay, latency | | Infrastructure | Prometheus + Grafana | QPS, latency, GPU util |

from evidently.report import Report
from evidently.metrics import DataDriftPreset

report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=ref_df, current_data=cur_df)
report.save_html("drift_report.html")

Data Science Patterns

from scipy import stats
import numpy as np

def ab_test(control: np.ndarray, treatment: np.ndarray, alpha: float = 0.05) -> dict:
    stat, p_value = stats.ttest_ind(control, treatment)
    lift = (treatment.mean() - control.mean()) / control.mean() * 100
    return {
        "p_value": float(p_value),
        "significant": p_value < alpha,
        "effect_size": float((treatment.mean() - control.mean()) / control.std()),
        "lift_pct": float(lift),
    }

Key Rules

1. Version data and models together (DVC or lakehouse time travel).
2. Pin Python, CUDA, and framework versions in Docker for reproducibility.
3. Log every experiment: parameters, metrics, artifacts, dataset hash.
4. Separate feature computation from model training. Feature store enables reuse.
5. Validate data quality before training and at inference time.
6. Use batch serving when latency allows; simpler and cheaper than online.
7. Monitor predictions vs actuals continuously. Trigger retraining on drift.
8. Never train on future data. Enforce temporal train/test splits.
9. Profile pipelines for skew, memory, and runtime before production.
10. Shadow-deploy new models alongside champion before switching traffic.