The Complete
Data Engineering
Playbook

Primary Source Categories

Key Repositories for ML

• Kaggle Datasets — kaggle.com/datasets — Community-driven, versioned, competition-tested
• Hugging Face Hub — huggingface.co/datasets — Best for NLP, multimodal; use datasets library
• TensorFlow Datasets (TFDS) — Standardised pipelines for CV/NLP benchmarks
• Paperswithcode Datasets — Tied to reproducible research benchmarks
• AWS Open Data Registry — Petabyte-scale scientific datasets on S3
• Google Dataset Search — datasetsearch.research.google.com
• OpenStreetMap / Overpass API — Geospatial ground truth

Data Gathering Strategy Framework

# Programmatic dataset acquisition with Hugging Face
from datasets import load_dataset
import pandas as pd
 
# Load a specific split and cache locally
ds = load_dataset("imdb", split="train", cache_dir="./data_cache")
 
# Convert to Pandas for exploration
df = ds.to_pandas()
print(df.dtypes)
print(df.describe(include='all'))
 
# For Kaggle API
# pip install kaggle
# Set KAGGLE_USERNAME and KAGGLE_KEY env variables
import subprocess
subprocess.run(["kaggle", "datasets", "download",
               "-d", "username/dataset-name",
               "--unzip", "-p", "./data"])
python

Data Lineage Tracking

Every dataset you gather should have a lineage record — a machine-readable provenance log documenting source URL, access date, licence, version hash, and the transformation chain applied. Tools like Apache Atlas, DataHub, and MLflow (for experiment context) help automate this.

# Simple lineage metadata pattern
import json, hashlib
from datetime import datetime
 
def record_lineage(source_url: str, local_path: str, licence: str) -> dict:
    with open(local_path, "rb") as f:
        sha256 = hashlib.sha256(f.read()).hexdigest()
    record = {
        "source": source_url,
        "local_path": local_path,
        "accessed_at": datetime.utcnow().isoformat(),
        "sha256": sha256,
        "licence": licence,
        "transformations": []
    }
    with open("lineage.json", "a") as f:
        f.write(json.dumps(record) + "\n")
    return record
python

Structuring Your Collection Architecture

A well-designed collection pipeline has three tiers: an ingestion layer (APIs, scrapers, streams), a landing zone (raw, immutable storage — think S3 or GCS), and a staging area (where format normalisation happens before the warehouse). Never write raw data directly to a transformed table.

File Format Selection

API Collection Pattern (REST & GraphQL)

import httpx, time, json
from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(stop=stop_after_attempt(5),
       wait=wait_exponential(multiplier=1, min=2, max=60))
async def fetch_paginated(url: str, headers: dict, params: dict) -> list:
    """
    Robust paginated API collection with exponential backoff.
    Handles rate-limiting (HTTP 429) gracefully.
    """
    results = []
    async with httpx.AsyncClient(timeout=30.0) as client:
        while url:
            resp = await client.get(url, headers=headers, params=params)
            if resp.status_code == 429:
                retry_after = int(resp.headers.get("Retry-After", 60))
                time.sleep(retry_after)
                continue
            resp.raise_for_status()
            data = resp.json()
            results.extend(data.get("items", []))
            url = data.get("next_page_url")   # pagination cursor
            params = {}  # cursor already encoded in next_page_url
    return results
python

Schema Validation with Great Expectations

import great_expectations as gx
 
context = gx.get_context()
ds = context.sources.add_pandas("my_source")
da = ds.add_dataframe_asset("users")
batch = da.build_batch_request()
 
suite = context.add_expectation_suite("users_suite")
validator = context.get_validator(batch_request=batch,
                                    expectation_suite_name="users_suite")
 
# Define schema contract
validator.expect_column_to_exist("user_id")
validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_be_unique("user_id")
validator.expect_column_values_to_be_between("age", min_value=0, max_value=120)
validator.expect_column_values_to_match_regex("email", r"^[^@]+@[^@]+\.[^@]+$")
 
results = validator.validate()
assert results.success, "Schema validation FAILED — pipeline halted."
python

Preparation: Sorting and Arranging

At this stage you perform schema alignment (unifying column names and types across sources), join strategy selection (star vs. snowflake schemas), and train/val/test split design. The order matters — you must design splits before any preprocessing that uses statistics from the data (like mean imputation), otherwise you leak test-set statistics into your training pipeline.

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer
from sklearn.model_selection import train_test_split
 
# Split FIRST
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)
 
numeric_pipe = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])
 
cat_pipe = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False))
])
 
preprocessor = ColumnTransformer([
    ("num", numeric_pipe, numeric_cols),
    ("cat", cat_pipe, categorical_cols)
])
 
# FIT on train only — TRANSFORM both
preprocessor.fit(X_train)
X_train_processed = preprocessor.transform(X_train)
X_test_processed  = preprocessor.transform(X_test)
python

Core Preprocessing Operations with Mathematical Intuition

1. Feature Scaling

Neural networks, SVMs, and K-Means are scale-sensitive. Tree-based models (XGBoost, Random Forest) are scale-invariant. Always understand your algorithm before scaling.

2. Missing Value Strategies

from sklearn.experimental import enable_iterative_imputer  # noqa
from sklearn.impute import IterativeImputer
from sklearn.linear_model import BayesianRidge
import numpy as np
 
imputer = IterativeImputer(
    estimator=BayesianRidge(),
    n_nearest_features=5,      # use 5 most correlated features
    max_iter=10,
    random_state=42,
    tol=1e-3
)
X_imputed = imputer.fit_transform(X_train)  # fit on train only!
python

3. Encoding Categorical Variables

4. Feature Engineering: Interaction Terms & Polynomial Features

5. Dimensionality Reduction

import numpy as np
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
 
# Determine optimal n_components
pca_full = PCA().fit(X_train_scaled)
cumvar = np.cumsum(pca_full.explained_variance_ratio_)
n_components = np.argmax(cumvar >= 0.95) + 1
print(f"Components for 95% variance: {n_components}")
 
pca = PCA(n_components=n_components, random_state=42)
X_reduced = pca.fit_transform(X_train_scaled)
python

Outlier Detection Methods

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
 
def comprehensive_outlier_report(df: pd.DataFrame, numeric_cols: list) -> pd.DataFrame:
    report = pd.DataFrame(index=numeric_cols)
    # Z-Score outliers
    z = (df[numeric_cols] - df[numeric_cols].mean()) / df[numeric_cols].std()
    report["zscore_outliers"] = (z.abs() > 3).sum()
    # IQR outliers
    Q1 = df[numeric_cols].quantile(0.25)
    Q3 = df[numeric_cols].quantile(0.75)
    IQR = Q3 - Q1
    iqr_mask = (df[numeric_cols] < (Q1 - 1.5*IQR)) | (df[numeric_cols] > (Q3 + 1.5*IQR))
    report["iqr_outliers"] = iqr_mask.sum()
    return report
 
# Isolation Forest for multivariate anomaly detection
iso = IsolationForest(contamination=0.05, random_state=42, n_jobs=-1)
outlier_labels = iso.fit_predict(X_numeric)  # -1 = anomaly, 1 = normal
df_clean = df[outlier_labels == 1]
print(f"Removed {(outlier_labels == -1).sum()} anomalies ({(outlier_labels == -1).mean():.1%})")
python

Duplicate Detection

# Exact duplicates
exact_dupes = df.duplicated(keep="first")
df = df[~exact_dupes]
 
# Fuzzy deduplication (near-duplicates in text)
from datasketch import MinHash, MinHashLSH
 
def minhash_dedup(texts: list, threshold: float = 0.8) -> list:
    """MinHash LSH for approximate deduplication — O(n) vs O(n²)"""
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    minhashes = {}
    keep_indices = []
 
    for i, text in enumerate(texts):
        m = MinHash(num_perm=128)
        for word in text.lower().split():
            m.update(word.encode("utf8"))
        if not lsh.query(m):  # no similar doc found
            lsh.insert(f"doc_{i}", m)
            keep_indices.append(i)
    return keep_indices
python

Data Manipulation Best Practices

Pandas — Production Patterns

import pandas as pd
import numpy as np
 
# Use vectorised operations — NEVER iterate rows
df["revenue_log"] = np.log1p(df["revenue"])  # log(1+x) handles zeros
 
# Downcasting dtypes saves 60-80% memory
def reduce_memory(df):
    for col in df.select_dtypes(include=["float64"]).columns:
        df[col] = pd.to_numeric(df[col], downcast="float")
    for col in df.select_dtypes(include=["int64"]).columns:
        df[col] = pd.to_numeric(df[col], downcast="integer")
    for col in df.select_dtypes(include=["object"]).columns:
        if df[col].nunique() / len(df) < 0.05:  # <5% unique = categorical
            df[col] = df[col].astype("category")
    return df
 
# Window functions for time-series features
df = df.sort_values(["user_id", "timestamp"])
df["rolling_7d_avg"] = (
    df.groupby("user_id")["value"]
    .transform(lambda x: x.rolling(7, min_periods=1).mean())
)
 
# Efficient merge strategy
df_merged = df_left.merge(
    df_right,
    on="id",
    how="left",
    validate="m:1"      # asserts join cardinality — catches duplicates
)
python

SQL Manipulation (PostgreSQL / DuckDB)

-- Window function: rolling 30-day revenue per user
SELECT
    user_id,
    event_date,
    revenue,
    SUM(revenue) OVER (
        PARTITION BY user_id
        ORDER BY event_date
        ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
    ) AS rolling_30d_revenue,
 
    -- Percentile rank within cohort
    PERCENT_RANK() OVER (PARTITION BY cohort_month ORDER BY revenue) AS pct_rank,
 
    -- Lag features for churn modelling
    LAG(revenue, 1) OVER (PARTITION BY user_id ORDER BY event_date) AS prev_revenue,
    revenue - LAG(revenue, 1) OVER (PARTITION BY user_id ORDER BY event_date) AS mom_delta
FROM transactions;
sql

The Ethics Checklist (Before Any Dataset Publication)

• Informed consent: was data collected with explicit user knowledge and agreement?
• Purpose limitation: are you using data only for the stated purpose?
• Data minimisation: have you collected the minimum data sufficient for the task?
• Anonymisation / pseudonymisation: are PII fields removed or hashed?
• Re-identification risk: is the combination of fields enough to re-identify individuals?
• Demographic representation: does your dataset reflect the population the model serves?
• Labelling fairness: were annotators from diverse backgrounds? Check inter-annotator agreement.
• Historical bias: does your target variable encode historical discrimination (e.g., loan defaults shaped by redlining)?
• Right to erasure: can you delete a user's data from the training set and retrain?
• Transparency: is the dataset documented with a Datasheet (Gebru et al. 2018) or Data Nutrition Label?

Bias Detection — Mathematical Framework

from fairlearn.metrics import MetricFrame, selection_rate, true_positive_rate
from fairlearn.reductions import ExponentiatedGradient, DemographicParity
import pandas as pd
 
# Compute fairness metrics across protected groups
mf = MetricFrame(
    metrics={
        "selection_rate": selection_rate,
        "tpr": true_positive_rate,
    },
    y_true=y_test,
    y_pred=y_pred,
    sensitive_features=X_test["gender"]  # protected attribute
)
print(mf.by_group)
print("Disparate Impact:", mf.difference())
 
# Bias mitigation with Exponentiated Gradient
constraint = DemographicParity()
mitigator = ExponentiatedGradient(base_estimator, constraint)
mitigator.fit(X_train, y_train, sensitive_features=X_train["gender"])
y_pred_fair = mitigator.predict(X_test)
python

PII Detection & Anonymisation

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
 
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
 
text = "John Smith's email is john@example.com and his phone is +1-555-0123"
results = analyzer.analyze(text=text, language="en")
anonymized = anonymizer.anonymize(text=text, analyzer_results=results)
print(anonymized.text)
# Output: "<PERSON>'s email is <EMAIL_ADDRESS> and phone is <PHONE_NUMBER>"
python

Data Governance Policy Framework

Data Classification Taxonomy

Dependency Management

# pyproject.toml — pin dependencies with hash verification
[tool.poetry.dependencies]
python = "^3.11"
pandas = "~2.2"      # minor version locked
numpy = "~1.26"
scikit-learn = "~1.4"
torch = {version = "~2.2", extras = ["cuda12"]}
 
# Generate lock file with reproducible hashes
# poetry lock --no-update
# pip-compile --generate-hashes requirements.in
toml

Data Poisoning Defences

Differential Privacy for ML

from opacus import PrivacyEngine
import torch
 
privacy_engine = PrivacyEngine()
model, optimizer, data_loader = privacy_engine.make_private_with_epsilon(
    module=model,
    optimizer=optimizer,
    data_loader=data_loader,
    epochs=50,
    target_epsilon=1.0,   # strong privacy
    target_delta=1e-5,
    max_grad_norm=1.0
)
print(f"σ = {optimizer.noise_multiplier:.3f}")
python

Memory Management Fundamentals

import pandas as pd
import numpy as np
from concurrent.futures import ProcessPoolExecutor
import psutil, gc
 
def process_chunk(chunk: pd.DataFrame) -> pd.DataFrame:
    """Pure function — safe for multiprocessing"""
    chunk = chunk.dropna(subset=["value"])
    chunk["log_value"] = np.log1p(chunk["value"])
    return chunk
 
def stream_process_csv(filepath: str, chunksize: int = 100_000) -> pd.DataFrame:
    """
    Memory-safe streaming CSV processing.
    Processes in chunks, collects results, forces GC between chunks.
    """
    results = []
    mem = psutil.virtual_memory()
    safe_chunk = int((mem.available * 0.3) / (50 * 8))  # estimate bytes per row
 
    for i, chunk in enumerate(pd.read_csv(filepath, chunksize=safe_chunk)):
        processed = process_chunk(chunk)
        results.append(processed)
        del chunk, processed
        gc.collect()  # explicit GC trigger
        if i % 10 == 0:
            print(f"Processed {i * safe_chunk:,} rows | "
                  f"Memory: {psutil.virtual_memory().percent:.1f}%")
 
    return pd.concat(results, ignore_index=True)
python

Apache Spark — DataFrame API Best Practices

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from pyspark.ml.feature import StandardScaler, VectorAssembler
 
spark = (SparkSession.builder
    .appName("FeatureEngineering")
    .config("spark.sql.adaptive.enabled", "true")     # AQE for skew handling
    .config("spark.sql.shuffle.partitions", "200")
    .config("spark.memory.offHeap.enabled", "true")
    .config("spark.memory.offHeap.size", "4g")
    .getOrCreate())
 
df = spark.read.parquet("s3://bucket/data/*.parquet")
 
# Partition pruning — predicate pushed to storage layer
df_filtered = df.filter((F.col("year") == 2024) & (F.col("region") == "APAC"))
 
# Window aggregation — distributed version of Pandas groupby
w = Window.partitionBy("user_id").orderBy("ts").rowsBetween(-29, 0)
df_feat = df_filtered.withColumn(
    "rolling_30d", F.sum("revenue").over(w)
)
 
# Cache hot DataFrames — only if reused multiple times
df_feat.cache().count()  # trigger materialisation
 
# Write back to Delta Lake (ACID)
df_feat.write.format("delta").mode("overwrite").save("s3://bucket/features/")
python

Kafka — Streaming Data Pipeline

from confluent_kafka import Consumer, Producer
import json
 
# Producer: ingest raw events
producer = Producer({"bootstrap.servers": "kafka:9092"})
 
def delivery_report(err, msg):
    if err:
        print(f"Delivery failed: {err}")
 
def publish_event(topic: str, key: str, value: dict):
    producer.produce(
        topic, key=key.encode(),
        value=json.dumps(value).encode(),
        callback=delivery_report
    )
    producer.poll(0)  # non-blocking flush
 
# Consumer: feature extraction from stream
consumer = Consumer({
    "bootstrap.servers": "kafka:9092",
    "group.id": "feature-pipeline",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False    # manual commit = at-least-once
})
consumer.subscribe(["raw-events"])
python

Structured Logging Pattern

import structlog, logging, sys
from functools import wraps
 
# Configure structured JSON logging
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    logger_factory=structlog.PrintLoggerFactory()
)
 
logger = structlog.get_logger()
 
def pipeline_step(step_name: str):
    """Decorator: auto-log entry/exit/error for pipeline stages."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            log = logger.bind(step=step_name, args_len=len(args))
            log.info("step.start")
            try:
                result = func(*args, **kwargs)
                log.info("step.success")
                return result
            except ValueError as e:
                log.error("step.validation_error", error=str(e), exc_info=True)
                raise
            except Exception as e:
                log.critical("step.fatal_error", error=str(e), exc_info=True)
                raise
        return wrapper
    return decorator
 
@pipeline_step("feature_normalisation")
def normalise_features(df):
    if df.empty:
        raise ValueError("Input DataFrame is empty")
    return (df - df.mean()) / df.std()
python

Data Quality Monitoring with Evidently

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
 
report = Report(metrics=[
    DataDriftPreset(),
    DataQualityPreset()
])
report.run(reference_data=df_train, current_data=df_production)
report.save_html("drift_report.html")
 
# Raise alert if drift detected
drift_result = report.as_dict()["metrics"][0]["result"]
if drift_result["dataset_drift"]:
    raise RuntimeError(f"Data drift detected: {drift_result['share_of_drifted_columns']:.0%} columns drifted")
python

NumPy — The Foundation

import numpy as np
 
# Einstein summation — express any tensor contraction
A = np.random.randn(100, 50)   # batch_size × features
B = np.random.randn(50, 30)   # features × hidden
C = np.einsum("bi,ih->bh", A, B)  # equivalent to A @ B
 
# Vectorised cosine similarity matrix (no loops)
def cosine_similarity_matrix(X: np.ndarray) -> np.ndarray:
    norms = np.linalg.norm(X, axis=1, keepdims=True)  # (n,1)
    X_norm = X / (norms + 1e-8)                            # broadcast
    return X_norm @ X_norm.T                                # (n,n)
 
# Memory-mapped arrays for out-of-core processing
mmap = np.memmap("large_array.npy", dtype=np.float32,
                 mode="r", shape=(10_000_000, 128))
batch = mmap[0:1000].copy()  # loads only this slice into RAM
python

Pandas — Advanced Patterns

import pandas as pd
 
# Method chaining — readable, pipe-based transformations
result = (
    pd.read_parquet("events.parquet")
    .pipe(reduce_memory)
    .query("event_type == 'purchase' and amount > 0")
    .assign(
        log_amount = lambda df: np.log1p(df["amount"]),
        hour_of_day = lambda df: df["timestamp"].dt.hour
    )
    .groupby(["user_id", pd.Grouper(key="timestamp", freq="1D")])
    .agg(daily_revenue=("amount", "sum"), n_purchases=("amount", "count"))
    .reset_index()
)
python

JAX — For High-Performance ML Research

import jax.numpy as jnp
from jax import grad, jit, vmap, random
 
# Automatic differentiation through any computation
def mse_loss(params, X, y):
    predictions = jnp.dot(X, params["w"]) + params["b"]
    return jnp.mean((predictions - y) ** 2)
 
grad_fn = jit(grad(mse_loss))  # compiled gradient function
grads = grad_fn(params, X_batch, y_batch)  # {w: dL/dw, b: dL/db}
 
# vmap: apply single-sample function to a batch (no for-loop!)
def predict_single(params, x):
    return jnp.dot(x, params)
 
predict_batch = vmap(predict_single, in_axes=(None, 0))  # params fixed, x batched
python

PyTorch — Production Data Loading

import torch
from torch.utils.data import Dataset, DataLoader
from torch.utils.data.distributed import DistributedSampler
 
class TabularDataset(Dataset):
    def __init__(self, X: np.ndarray, y: np.ndarray):
        self.X = torch.tensor(X, dtype=torch.float32)
        self.y = torch.tensor(y, dtype=torch.long)
 
    def __len__(self): return len(self.X)
    def __getitem__(self, idx): return self.X[idx], self.y[idx]
 
# Production DataLoader with pinned memory for GPU transfer
loader = DataLoader(
    TabularDataset(X_train, y_train),
    batch_size=512,
    num_workers=4,        # parallel data loading workers
    pin_memory=True,       # faster CPU→GPU transfer
    persistent_workers=True, # avoid re-spawning workers each epoch
    prefetch_factor=2,    # pre-fetch 2 batches ahead
    sampler=DistributedSampler(dataset) # for multi-GPU DDP
)
python

TensorFlow / Keras — Production Data Pipelines

TensorFlow's tf.data API is the gold standard for building GPU-saturating input pipelines. The key is to overlap data loading (I/O-bound) with model execution (compute-bound) using prefetching and parallel map.

import tensorflow as tf
import numpy as np

# Production tf.data pipeline — saturate GPU with parallel I/O
def build_training_pipeline(
    file_pattern: str,
    batch_size: int = 512,
    num_parallel_reads: int = 8,
    shuffle_buffer: int = 10_000
) -> tf.data.Dataset:
    """High-performance TFRecord pipeline with:
    - Parallel file interleaving
    - Prefetch with AUTOTUNE
    - Cached in-memory after first epoch
    """
    feature_spec = {
        "features": tf.io.FixedLenFeature([128], tf.float32),
        "label": tf.io.FixedLenFeature([], tf.int64),
    }

    def parse_example(serialized):
        parsed = tf.io.parse_single_example(serialized, feature_spec)
        return parsed["features"], parsed["label"]

    files = tf.data.Dataset.list_files(file_pattern, shuffle=True)

    ds = files.interleave(
        lambda f: tf.data.TFRecordDataset(f, compression_type="GZIP"),
        num_parallel_calls=tf.data.AUTOTUNE,
        cycle_length=num_parallel_reads,
        deterministic=False  # non-deterministic for speed
    )

    ds = (
        ds
        .shuffle(shuffle_buffer)
        .map(parse_example, num_parallel_calls=tf.data.AUTOTUNE)
        .batch(batch_size, drop_remainder=True)  # drop_remainder for TPU
        .cache()             # cache after first epoch (fits in RAM)
        .prefetch(tf.data.AUTOTUNE)  # overlap data prep with training
    )
    return ds

# Keras model with mixed precision (2× speed on modern GPUs)
tf.keras.mixed_precision.set_global_policy("mixed_float16")

model = tf.keras.Sequential([
    tf.keras.layers.Dense(256, activation="relu", input_shape=(128,)),
    tf.keras.layers.BatchNormalization(),
    tf.keras.layers.Dropout(0.3),
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dense(1, activation="sigmoid", dtype="float32")
])

# Compile with TF's built-in AUC metric for imbalanced data
model.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
    loss="binary_crossentropy",
    metrics=[tf.keras.metrics.AUC(name="auroc")]
)

train_ds = build_training_pipeline("gs://bucket/train/*.tfrecord")
model.fit(train_ds, epochs=50, callbacks=[
    tf.keras.callbacks.EarlyStopping(monitor="val_auroc", patience=5, mode="max"),
    tf.keras.callbacks.ModelCheckpoint("best_model.keras", save_best_only=True)
])

# Export to SavedModel for TF Serving / TFLite
model.export("saved_model/fraud_detector")
python

When to Use Which Tool

Crawlee (Node.js) — Enterprise Scraping

// Crawlee with Playwright for JS-heavy sites
import { PlaywrightCrawler, Dataset } from 'crawlee';
 
const crawler = new PlaywrightCrawler({
    maxRequestsPerCrawl: 1000,
    maxConcurrency: 5,
    requestHandlerTimeoutSecs: 30,
 
    async requestHandler({ page, request, enqueueLinks }) {
        // Wait for dynamic content
        await page.waitForSelector('.product-card', { timeout: 10000 });
 
        const items = await page.$$eval('.product-card', cards =>
            cards.map(c => ({
                title: c.querySelector('h2')?.textContent,
                price: c.querySelector('.price')?.textContent,
                url: c.querySelector('a')?.href
            }))
        );
        await Dataset.pushData(items);
        await enqueueLinks({ selector: 'a.next-page' });
    },
    failedRequestHandler: ({ request }) =>
        console.error(`Failed: ${request.url}`)
});
 
await crawler.run(['https://example.com/products']);
javascript

Crawl4AI — LLM-Powered Extraction (Python)

from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction_strategy import LLMExtractionStrategy
import asyncio, json
 
async def extract_structured_data(url: str):
    strategy = LLMExtractionStrategy(
        provider="openai/gpt-4o-mini",
        api_token="YOUR_KEY",
        schema={
            "type": "object",
            "properties": {
                "company_name": {"type": "string"},
                "founding_year": {"type": "integer"},
                "revenue": {"type": "number"}
            }
        },
        instruction="Extract company details from the page."
    )
    async with AsyncWebCrawler(verbose=True) as crawler:
        result = await crawler.arun(url=url, extraction_strategy=strategy)
        return json.loads(result.extracted_content)
 
asyncio.run(extract_structured_data("https://example.com/about"))
python

Puppeteer — Headless Browser Scraping (Node.js)

// Puppeteer with stealth plugin — handles SPAs and bot detection
import puppeteer from 'puppeteer-extra';
import StealthPlugin from 'puppeteer-extra-plugin-stealth';
import fs from 'fs/promises';

puppeteer.use(StealthPlugin());

async function scrapeProducts(url) {
    const browser = await puppeteer.launch({
        headless: 'new',        // new headless mode (Chrome 112+)
        args: [
            '--no-sandbox',
            '--disable-setuid-sandbox',
            '--disable-dev-shm-usage'   // for Docker
        ]
    });

    const page = await browser.newPage();
    await page.setViewport({ width: 1280, height: 800 });

    // Navigate and wait for dynamic content
    await page.goto(url, { waitUntil: 'networkidle0', timeout: 30000 });

    // Auto-scroll to trigger lazy-loaded content
    await page.evaluate(async () => {
        await new Promise(resolve => {
            let totalHeight = 0;
            const distance = 300;
            const timer = setInterval(() => {
                window.scrollBy(0, distance);
                totalHeight += distance;
                if (totalHeight >= document.body.scrollHeight) {
                    clearInterval(timer);
                    resolve();
                }
            }, 200);
        });
    });

    // Extract structured data
    const products = await page.$$eval('.product-item', items =>
        items.map(el => ({
            name: el.querySelector('h3')?.textContent?.trim(),
            price: parseFloat(el.querySelector('.price')?.textContent?.replace(/[^0-9.]/g, '')),
            rating: parseFloat(el.querySelector('[data-rating]')?.getAttribute('data-rating')),
            url: el.querySelector('a')?.href
        }))
    );

    await fs.writeFile('products.json', JSON.stringify(products, null, 2));
    console.log(`Scraped ${products.length} products`);
    await browser.close();
    return products;
}

scrapeProducts('https://example.com/shop');
javascript

Playwright — Anti-Bot Techniques

from playwright.async_api import async_playwright
import asyncio, random
 
async def stealth_scrape(url: str):
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=['--disable-blink-features=AutomationControlled']
        )
        context = await browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64)...",
            viewport={"width": 1280, "height": 800},
            locale="en-US"
        )
        # Mask navigator.webdriver fingerprint
        await context.add_init_script(
            "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
        )
        page = await context.new_page()
        await page.goto(url, wait_until="networkidle")
        # Human-like delay
        await asyncio.sleep(random.uniform(1.5, 3.5))
        content = await page.content()
        await browser.close()
        return content
python

from sdv.single_table import CTGANSynthesizer
from sdv.metadata import SingleTableMetadata
from sdv.evaluation.single_table import run_diagnostic, evaluate_quality
 
metadata = SingleTableMetadata()
metadata.detect_from_dataframe(df_real)
metadata.update_column("user_id", sdtype="id")        # mark as ID, not feature
metadata.update_column("churn", sdtype="categorical")  # target column
 
synthesizer = CTGANSynthesizer(
    metadata,
    epochs=300,
    batch_size=500,
    discriminator_steps=1,
    verbose=True
)
synthesizer.fit(df_real)
df_synthetic = synthesizer.sample(num_rows=50_000)
 
# Evaluate quality: column shapes + correlation
quality = evaluate_quality(df_real, df_synthetic, metadata)
print(f"Quality Score: {quality.get_score():.2f}")  # aim for >0.85
python

LLM-Powered Synthetic NLP Dataset Pipeline

import anthropic, json
from typing import Iterator
 
client = anthropic.Anthropic()
 
def generate_synthetic_samples(
    task: str, labels: list, n_per_label: int = 100
) -> Iterator[dict]:
    """Generate labelled NLP training samples via Claude."""
    for label in labels:
        prompt = f"""Generate {n_per_label} diverse text samples for a {task} classifier.
Label: {label}
Requirements:
- Vary writing style (formal, casual, terse, verbose)
- Vary sentiment polarity where applicable
- Avoid exact duplicates
Return ONLY a JSON array of strings, no extra text."""
        msg = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}]
        )
        samples = json.loads(msg.content[0].text)
        for text in samples:
            yield {"text": text, "label": label}
python

The Verdict — ML Engineer Stack 2026

import duckdb
import pandas as pd
 
# DuckDB — the ML engineer's Swiss army knife
con = duckdb.connect("ml_features.duckdb")
 
# Read Parquet directly — no loading into memory first!
result = con.execute("""
    SELECT
        user_id,
        AVG(amount) FILTER (WHERE event_type='purchase') AS avg_purchase,
        COUNT(*) FILTER (WHERE event_type='click')       AS click_count,
        APPROX_QUANTILE(amount, 0.95)                    AS p95_amount
    FROM read_parquet('s3://bucket/events/*.parquet')
    WHERE YEAR(event_date) = 2024
    GROUP BY user_id
    HAVING COUNT(*) > 10
""").df()
 
# Write feature table back to parquet
con.execute("COPY result TO 'features.parquet' (FORMAT PARQUET)")
python

Airflow DAG for ML Retraining Pipeline

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime, timedelta
 
default_args = {
    "owner": "ml-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["ml-alerts@company.com"]
}
 
with DAG(
    "weekly_churn_retrain",
    default_args=default_args,
    schedule_interval="0 2 * * 1",  # every Monday at 2am
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
 
    validate_data = PythonOperator(
        task_id="validate_input_data",
        python_callable=lambda: run_great_expectations_suite("churn_suite")
    )
 
    feature_eng = SparkSubmitOperator(
        task_id="compute_features",
        application="s3://scripts/feature_engineering.py",
        conf={"spark.executor.memory": "8g", "spark.executor.cores": "4"}
    )
 
    train_model = PythonOperator(
        task_id="train_xgboost",
        python_callable=train_and_log_to_mlflow
    )
 
    validate_model = PythonOperator(
        task_id="validate_model_metrics",
        python_callable=compare_against_champion
    )
 
    deploy = PythonOperator(
        task_id="deploy_to_sagemaker",
        python_callable=promote_challenger_to_champion
    )
 
    validate_data >> feature_eng >> train_model >> validate_model >> deploy
python

Essential Books

• Book
  Designing Machine Learning Systems — Chip Huyen (O'Reilly 2022) — The definitive guide to production ML, covers feature stores, data quality, and deployment end-to-end.
• Book
  Fundamentals of Data Engineering — Reis & Housley (O'Reilly 2022) — Best overview of the full data stack from ingestion to serving.
• Book
  High Performance Spark — Karau & Warren (O'Reilly) — Internals, tuning, and optimisation of PySpark pipelines.
• Book
  The Elements of Statistical Learning — Hastie, Tibshirani, Friedman (Free PDF: web.stanford.edu/~hastie/ElemStatLearn/) — Mathematical foundations essential for every ML engineer.
• Book
  Data Pipelines Pocket Reference — Densmore (O'Reilly 2021) — Practical guide covering Kafka, Airflow, Spark.
• Book
  Responsible Data Science — Schelter & Stoyanovich (MIT Press) — Ethics, fairness, and governance with code examples.

Landmark Papers

• Paper
  Datasheets for Datasets — Gebru et al. (2018) — arxiv.org/abs/1803.09010 — Standard for dataset documentation. Read before publishing any dataset.
• Paper
  Data Management Challenges in Production ML — Sculley et al. (NIPS 2015) — "Hidden Technical Debt in ML Systems" — Essential reading on ML engineering pitfalls.
• Paper
  Differential Privacy: A Survey of Results — Dwork (2008) — Foundational DP theory with mathematical proofs.
• Paper
  SMOTE: Synthetic Minority Over-sampling Technique — Chawla et al. (JAIR 2002) — Original SMOTE paper, still the most cited class imbalance technique.
• Paper
  Attention Is All You Need — Vaswani et al. (2017) — The transformer architecture that powers every modern NLP feature pipeline.
• Paper
  Lakehouse: A New Generation of Open Platforms — Zaharia et al. (CIDR 2021) — Foundational paper for Delta Lake / Iceberg / Hudi architectures.

Courses & Learning Paths

• Course
  mlops-zoomcamp — github.com/DataTalksClub/mlops-zoomcamp — Free, hands-on MLOps with MLflow, Prefect, and FastAPI. Highest ROI course in this space.
• Course
  Full Stack Deep Learning — fullstackdeeplearning.com — From data pipelines to LLM deployment. Free online lectures.
• Course
  Stanford CS329S: ML Systems Design — stanford-cs329s.github.io — Chip Huyen's course. Lecture notes freely available.
• Course
  dbt Fundamentals — courses.getdbt.com — Free, the best way to learn dbt for data transformation.
• Docs
  Apache Spark Official Documentation — spark.apache.org/docs/latest — Internals, tuning guide, and PySpark API.
• Docs
  DuckDB Documentation — duckdb.org/docs — Especially the SQL extensions and Parquet/Arrow integration.