← Back to all products

ML Pipeline Templates

$49

End-to-end ML pipelines: data ingestion, preprocessing, training, evaluation, and deployment orchestration.

📁 18 files🏷 v1.0.0
JSONMarkdownPythonYAMLAirflow

📄 Product Preview

Try the interactive reader and demo tools below, or get the full product with all content unlocked.

📖 Interactive Reader (Free Preview) ⚙ Try Demo Tools 📦 Download Free Sample

📁 File Structure 18 files

ml-pipeline-templates/ ├── LICENSE ├── README.md ├── configs/ │ └── pipeline.yaml ├── examples/ │ ├── run_pipeline.py │ └── sklearn_pipeline.py ├── guides/ │ ├── orchestration-options.md │ └── pipeline-architecture.md ├── requirements.txt ├── src/ │ └── ml_pipeline/ │ ├── __init__.py │ ├── evaluation.py │ ├── ingestion.py │ ├── pipeline.py │ ├── preprocessing.py │ ├── steps.py │ └── training.py └── tests/ ├── test_pipeline.py └── test_preprocessing.py

📖 Documentation Preview README excerpt

ML Pipeline Templates

End-to-end, dependency-light templates for building tabular machine-learning

pipelines: data ingestion -> preprocessing -> training -> evaluation, chained

together by a small DAG orchestrator you can actually read.

The orchestration core is pure standard library, so it is fast to import and

trivial to test. The ML stages use pandas/numpy and prefer scikit-learn when

it is installed -- but every stage has a numpy fallback, so the whole pipeline

(and the test suite) runs end-to-end even with nothing but pandas + numpy.

Features

  • DAG orchestrator (Pipeline / Step) -- declare dependencies, get

deterministic topological execution, per-step timing/status, cycle and

missing-dependency detection, and fail_fast handling. No third-party deps.

  • Ingestion into one Dataset interface from CSV, Parquet, or SQL (or an

in-memory DataFrame), so downstream stages never touch a file path.

  • Preprocessing built on a scikit-learn ColumnTransformer (impute + scale +

one-hot encode), with a dependency-free PandasFeaturePipeline fallback and a

leakage-safe train/test split (optionally stratified).

  • Training with backend-agnostic K-fold cross-validation that works with any

fit/predict estimator -- scikit-learn models or the bundled numpy

baselines (RidgeRegressor, MeanRegressor, MajorityClassifier).

  • Evaluation with pure-numpy regression and classification metrics plus a

renderable EvaluationReport (Markdown, including a confusion matrix).

  • Config-driven -- build the standard four-stage pipeline from

configs/pipeline.yaml; retarget your data without editing Python.

  • Tested -- a pytest suite covers DAG ordering, context passing, error

handling, and the preprocessing transforms.

Requirements

  • Python 3.10+
  • pandas, numpy, PyYAML (required by the ML stages)
  • scikit-learn (recommended; auto-used when present), pyarrow (for Parquet) --

both optional


pip install -r requirements.txt

Quick start

Everything runs straight from the product folder -- the examples and tests add

src/ to the path, so no install step is needed.


# 1) Run the full pipeline on a generated toy dataset
python examples/run_pipeline.py

# 2) Run it from the YAML config instead of an inline config
python examples/run_pipeline.py --config configs/pipeline.yaml

# 3) See the scikit-learn ColumnTransformer + RandomForest path
python examples/sklearn_pipeline.py        # prints an install hint if sklearn is absent

# 4) Run the test suite
pip install pytest
python -m pytest tests/ -v

*... continues with setup instructions, usage examples, and more.*

📄 Code Sample .py preview

src/ml_pipeline/evaluation.py """Metrics and evaluation reports, implemented with numpy only. Regression and classification metrics are computed directly from numpy arrays so that reports can be produced in any environment, with or without scikit-learn. The formulas match the standard definitions used by ``sklearn.metrics`` (verified in ``tests/``), so swapping in ``sklearn.metrics`` later changes nothing material. """ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path from typing import Any import numpy as np # --------------------------------------------------------------------------- # # Task inference # --------------------------------------------------------------------------- # def infer_task(y: Any) -> str: """Guess whether a target vector represents ``"regression"`` or ``"classification"``. Heuristic: float targets with many distinct values are regression; integer, boolean, string, or low-cardinality targets are classification. """ arr = np.asarray(y) if arr.dtype.kind in {"U", "S", "O", "b"}: return "classification" distinct = np.unique(arr[~_nan_mask(arr)]) if arr.dtype.kind == "f" else np.unique(arr) if arr.dtype.kind == "f": # All-integer floats with low cardinality look like encoded class labels. is_integral = np.all(np.isfinite(arr)) and np.allclose(arr, np.round(arr)) if is_integral and len(distinct) <= max(20, int(0.05 * len(arr))): return "classification" return "regression" # Integer dtype: low cardinality -> classification, otherwise regression. if len(distinct) <= max(20, int(0.05 * len(arr))): return "classification" return "regression" def _nan_mask(arr: np.ndarray) -> np.ndarray: if arr.dtype.kind == "f": return np.isnan(arr) return np.zeros(arr.shape, dtype=bool) # --------------------------------------------------------------------------- # # ... 274 more lines ...
Buy Now — $49 Back to Products