{"id":853,"date":"2026-05-05T06:11:37","date_gmt":"2026-05-04T22:11:37","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=853"},"modified":"2026-05-05T06:11:37","modified_gmt":"2026-05-04T22:11:37","slug":"how-to-build-an-end-to-end-production-grade-machine-learning-pipeline-with-zenml-including-custom-materializers-metadata-tracking-and-hyperparameter-optimization","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=853","title":{"rendered":"How to Build an End-to-End Production Grade Machine Learning Pipeline with ZenML, Including Custom Materializers, Metadata Tracking, and Hyperparameter Optimization"},"content":{"rendered":"

In this tutorial, we walk through an end-to-end implementation of an advanced machine learning pipeline using ZenML<\/strong><\/a>. We begin by setting up the environment and initializing a ZenML project, then define a custom materializer that enables seamless serialization and metadata extraction for a domain-specific dataset object. As we progress, we build a modular pipeline that performs data loading, preprocessing, and a fan-out hyperparameter search across multiple models. We evaluate each candidate, log rich metadata at every step, and use a fan-in strategy to select and promote the best-performing model. Throughout the process, we leverage ZenML\u2019s model control plane, artifact tracking, and caching mechanisms to ensure full reproducibility, transparency, and efficiency.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
import os, sys, subprocess, json, shutil\nfrom pathlib import Path\n\n\ndef _sh(cmd, check=True):\n   print(f\"$ {' '.join(cmd)}\")\n   return subprocess.run(cmd, check=check)\n\n\n_sh([sys.executable, \"-m\", \"pip\", \"install\", \"-q\",\n    \"zenml[server]\", \"scikit-learn\", \"pandas\", \"pyarrow\"])\n\n\nPROJECT = Path(\"\/content\/zenml_advanced_tutorial\") if Path(\"\/content\").exists() \n   else Path.cwd() \/ \"zenml_advanced_tutorial\"\nif PROJECT.exists():\n   shutil.rmtree(PROJECT)\nPROJECT.mkdir(parents=True)\nos.chdir(PROJECT)\n\n\nos.environ[\"ZENML_ANALYTICS_OPT_IN\"] = \"false\"\nos.environ[\"ZENML_LOGGING_VERBOSITY\"] = \"WARN\"\n_sh([\"zenml\", \"init\"], check=False)<\/code><\/pre>\n<\/div>\n<\/div>\n
We set up the entire environment by installing required libraries and initializing a ZenML project workspace. We create a clean working directory and configure environment variables to control logging and analytics behavior. Finally, we bootstrap the ZenML repository so that all subsequent pipeline operations are properly tracked and managed.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
from typing import Annotated, Tuple, Dict, List, Any\nimport numpy as np\nfrom sklearn.datasets import load_breast_cancer\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier\nfrom sklearn.linear_model import LogisticRegression\nfrom sklearn.metrics import accuracy_score, f1_score, roc_auc_score\nfrom sklearn.preprocessing import StandardScaler\n\n\nfrom zenml import pipeline, step, log_metadata, Model, get_step_context\nfrom zenml.client import Client\nfrom zenml.materializers.base_materializer import BaseMaterializer\nfrom zenml.enums import ArtifactType\nfrom zenml.io import fileio\n\n\nclass DatasetBundle:\n   def __init__(self, X, y, feature_names, stats=None):\n       self.X = np.asarray(X)\n       self.y = np.asarray(y)\n       self.feature_names = list(feature_names)\n       self.stats = stats or {}\n\n\nclass DatasetBundleMaterializer(BaseMaterializer):\n   ASSOCIATED_TYPES = (DatasetBundle,)\n   ASSOCIATED_ARTIFACT_TYPE = ArtifactType.DATA\n\n\n   def load(self, data_type):\n       with fileio.open(os.path.join(self.uri, \"X.npy\"), \"rb\") as f:\n           X = np.load(f)\n       with fileio.open(os.path.join(self.uri, \"y.npy\"), \"rb\") as f:\n           y = np.load(f)\n       with fileio.open(os.path.join(self.uri, \"meta.json\"), \"r\") as f:\n           meta = json.loads(f.read())\n       return DatasetBundle(X, y, meta[\"feature_names\"], meta[\"stats\"])\n\n\n   def save(self, bundle):\n       with fileio.open(os.path.join(self.uri, \"X.npy\"), \"wb\") as f:\n           np.save(f, bundle.X)\n       with fileio.open(os.path.join(self.uri, \"y.npy\"), \"wb\") as f:\n           np.save(f, bundle.y)\n       with fileio.open(os.path.join(self.uri, \"meta.json\"), \"w\") as f:\n           f.write(json.dumps({\n               \"feature_names\": bundle.feature_names,\n               \"stats\": bundle.stats,\n           }))\n\n\n   def extract_metadata(self, bundle):\n       classes, counts = np.unique(bundle.y, return_counts=True)\n       return {\n           \"n_samples\": int(bundle.X.shape[0]),\n           \"n_features\": int(bundle.X.shape[1]),\n           \"class_distribution\": {str(c): int(n) for c, n in zip(classes, counts)},\n       }<\/code><\/pre>\n<\/div>\n<\/div>\n
We import all necessary libraries and define a custom data container along with its materializer. We implement logic to save, load, and extract metadata from our dataset, enabling seamless artifact handling in ZenML. This ensures that our data is not only stored efficiently but also enriched with meaningful, queryable metadata.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
@step(enable_cache=True)\ndef load_data() -> Annotated[DatasetBundle, \"raw_dataset\"]:\n   data = load_breast_cancer()\n   return DatasetBundle(\n       data.data, data.target, data.feature_names,\n       stats={\"source\": \"sklearn.datasets.load_breast_cancer\"},\n   )\n\n\n@step\ndef split_and_scale(\n   bundle: DatasetBundle,\n   test_size: float = 0.2,\n   random_state: int = 42,\n) -> Tuple[\n   Annotated[np.ndarray, \"X_train\"],\n   Annotated[np.ndarray, \"X_test\"],\n   Annotated[np.ndarray, \"y_train\"],\n   Annotated[np.ndarray, \"y_test\"],\n]:\n   X_tr, X_te, y_tr, y_te = train_test_split(\n       bundle.X, bundle.y, test_size=test_size,\n       random_state=random_state, stratify=bundle.y,\n   )\n   scaler = StandardScaler().fit(X_tr)\n   X_tr, X_te = scaler.transform(X_tr), scaler.transform(X_te)\n   log_metadata(metadata={\"train_size\": len(X_tr), \"test_size\": len(X_te)})\n   return X_tr, X_te, y_tr, y_te\n\n\n@step\ndef train_candidate(\n   X_train: np.ndarray,\n   y_train: np.ndarray,\n   model_type: str = \"random_forest\",\n   n_estimators: int = 100,\n   max_depth: int = 5,\n) -> Annotated[Any, \"candidate_model\"]:\n   if model_type == \"random_forest\":\n       m = RandomForestClassifier(n_estimators=n_estimators,\n                                  max_depth=max_depth, random_state=42)\n   elif model_type == \"gradient_boosting\":\n       m = GradientBoostingClassifier(n_estimators=n_estimators,\n                                      max_depth=max_depth, random_state=42)\n   else:\n       m = LogisticRegression(max_iter=2000, random_state=42)\n   m.fit(X_train, y_train)\n   log_metadata(metadata={\n       \"model_type\": model_type,\n       \"hyperparameters\": {\"n_estimators\": n_estimators, \"max_depth\": max_depth},\n   })\n   return m<\/code><\/pre>\n<\/div>\n<\/div>\n
We define core pipeline steps for loading data, splitting it, scaling features, and training model candidates. We ensure that data loading is cached for efficiency while logging key metadata during preprocessing and training. This forms the backbone of our pipeline, where each model is trained independently with its respective configuration.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
@step\ndef evaluate_candidate(\n   model: Any,\n   X_test: np.ndarray,\n   y_test: np.ndarray,\n   label: str,\n) -> Annotated[Dict[str, Any], \"metrics\"]:\n   preds = model.predict(X_test)\n   probs = (model.predict_proba(X_test)[:, 1]\n            if hasattr(model, \"predict_proba\") else preds)\n   metrics: Dict[str, Any] = {\n       \"accuracy\": float(accuracy_score(y_test, preds)),\n       \"f1\":       float(f1_score(y_test, preds)),\n       \"roc_auc\":  float(roc_auc_score(y_test, probs)),\n       \"label\":    label,\n   }\n   log_metadata(metadata=metrics)\n   return metrics\n\n\n@step\ndef select_best(\n   metrics_list: List[Dict[str, Any]],\n   models: List[Any],\n) -> Annotated[Any, \"production_model\"]:\n   best_idx = max(range(len(metrics_list)),\n                  key=lambda i: metrics_list[i][\"roc_auc\"])\n   best = metrics_list[best_idx]\n\n\n   ctx = get_step_context()\n   try:\n       ctx.model.log_metadata({\"chosen_candidate\": best,\n                               \"candidate_index\": best_idx})\n   except Exception as e:\n       print(f\"  (model metadata log skipped: {e})\")\n\n\n   log_metadata(metadata={\n       \"winning_metrics\": {k: v for k, v in best.items() if k != \"label\"},\n   })\n   print(f\"n\"\ud83c\udfc6\"  Best candidate: {best['label']}  \u2192  \"\n         f\"ROC AUC = {best['roc_auc']:.4f}n\")\n   return models[best_idx]<\/code><\/pre>\n<\/div>\n<\/div>\n
We evaluate each trained model using multiple performance metrics and log the results. We then implement a selection mechanism that identifies the best-performing model based on ROC AUC. Additionally, we attach relevant metadata to the model version, enabling traceability and informed decision-making.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
SEARCH_SPACE = [\n   {\"model_type\": \"random_forest\",     \"n_estimators\": 50,  \"max_depth\": 3},\n   {\"model_type\": \"random_forest\",     \"n_estimators\": 200, \"max_depth\": 7},\n   {\"model_type\": \"gradient_boosting\", \"n_estimators\": 100, \"max_depth\": 3},\n   {\"model_type\": \"logistic\",          \"n_estimators\": 1,   \"max_depth\": 1},\n]\n\n\nPRODUCTION_MODEL = Model(\n   name=\"breast_cancer_classifier\",\n   description=\"Best model from in-pipeline hyperparameter search\",\n   tags=[\"tutorial\", \"advanced\"],\n)\n\n\n@pipeline(model=PRODUCTION_MODEL, enable_cache=True)\ndef training_pipeline(test_size: float = 0.2):\n   bundle = load_data()\n\n\n   models, metrics = [], []\n   for i, cfg in enumerate(SEARCH_SPACE):\n       m = train_candidate(\n           X_train, y_train, **cfg,\n           id=f\"train_{i}_{cfg['model_type']}\",\n       )\n       s = evaluate_candidate(\n           m, X_test, y_test,\n           label=f\"{cfg['model_type']}(n={cfg['n_estimators']},d={cfg['max_depth']})\",\n           id=f\"eval_{i}\",\n       )\n       models.append(m)\n       metrics.append(s)\n\n\n   select_best(metrics, models)\n\n\nprint(\"n\" + \"=\" * 70 + \"n  RUNNING TRAINING PIPELINEn\" + \"=\" * 70)\nrun_obj = training_pipeline()\n\n\nprint(\"n\" + \"=\" * 70 + \"n  INSPECTING THE RUNn\" + \"=\" * 70)\nclient = Client()\nrun = client.get_pipeline_run(run_obj.id)\n\n\nprint(f\"nPipeline:   {run.pipeline.name}\")\nprint(f\"Run name:   {run.name}\")\nprint(f\"Status:     {run.status}\")\nprint(f\"Step runs:  {len(run.steps)}\")\nfor name, step_run in run.steps.items():\n   print(f\"  \u2022 {name:35s} status={step_run.status}\")\n\n\nprint(\"nRun-level metadata (aggregated from steps):\")\nfor k, v in (run.run_metadata or {}).items():\n   short = str(v)\n   print(f\"  {k}: {short[:80]}{'\u2026' if len(short) > 80 else ''}\")\n\n\nprint(\"n\" + \"-\" * 70 + \"n  MODEL CONTROL PLANEn\" + \"-\" * 70)\ntry:\n   mv = client.get_model_version(PRODUCTION_MODEL.name, \"latest\")\nexcept Exception:\n   mv = client.list_model_versions(model_name_or_id=PRODUCTION_MODEL.name)[0]\n\n\nprint(f\"Model:           {mv.model.name}\")\nprint(f\"Version:         {mv.name} (number={mv.number})\")\nlinked = list(mv.data_artifact_ids.keys()) if hasattr(mv, \"data_artifact_ids\") else []\nprint(f\"Linked outputs:  {linked or '(see dashboard)'}\")\nif mv.run_metadata:\n   print(\"Version metadata:\")\n   for k, v in dict(mv.run_metadata).items():\n       print(f\"  {k}: {str(v)[:80]}\")\n\n\nprint(\"n\" + \"-\" * 70 + \"n  RELOADING ARTIFACTS DIRECTLYn\" + \"-\" * 70)\nprod_artifact = client.get_artifact_version(\"production_model\")\nprod_model = prod_artifact.load()\nprint(f\"Loaded model class:   {type(prod_model).__name__}\")\nprint(f\"Artifact metadata:    {dict(prod_artifact.run_metadata) if prod_artifact.run_metadata else '{}'}\"[:120])\n\n\nX_test_arr = client.get_artifact_version(\"X_test\").load()\ny_test_arr = client.get_artifact_version(\"y_test\").load()\nacc = accuracy_score(y_test_arr, prod_model.predict(X_test_arr))\nprint(f\"Sanity-check accuracy on stored X_test: {acc:.4f}\")\n\n\nds_artifact = client.get_artifact_version(\"raw_dataset\")\nprint(f\"nraw_dataset auto-extracted metadata:\")\nfor k, v in (ds_artifact.run_metadata or {}).items():\n   print(f\"  {k}: {v}\")\n\n\nprint(\"n\" + \"=\" * 70 + \"n  RE-RUNNING \u2014 STEPS SHOULD BE CACHEDn\" + \"=\" * 70)\ntraining_pipeline()\n\n\nprint(\"\"\"\n\"\u2705\" Tutorial complete.\n\n\nWhat just happened:\n \u2022 Custom materializer serialized a domain object + auto-extracted metadata.\n \u2022 Fan-out: 4 candidates trained + evaluated as 8 distinct step runs.\n \u2022 Fan-in: select_best joined them and promoted the winner.\n \u2022 Model Control Plane created a versioned 'breast_cancer_classifier'.\n \u2022 Every artifact, metric, and hyperparameter was logged and queryable.\n \u2022 Second run hit the cache \u2014 zero recomputation.\n\n\nExplore further from this same Python session:\n Client().list_pipeline_runs()\n Client().list_model_versions(model_name_or_id=\"breast_cancer_classifier\")\n Client().list_artifact_versions(name=\"metrics\")\n\"\"\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the full pipeline, execute it, and inspect the results using the ZenML Client API. We perform a fan-out over multiple configurations, followed by a fan-in step to select the best model. Finally, we demonstrate artifact reuse, metadata inspection, and caching behavior by re-running the pipeline without redundant computation.<\/p>\n
In conclusion, we constructed a robust, production-style ML pipeline that demonstrates the full power of ZenML\u2019s orchestration capabilities. We observed how custom materializers enrich artifacts with meaningful metadata, how multiple model candidates can be trained and evaluated in parallel, and how the best model is automatically selected and versioned. We also explored how to inspect pipeline runs, retrieve artifacts directly without recomputation, and verify model performance using stored data. Also, we saw caching in action during a re-run, confirming that redundant computations are avoided. This workflow provides a strong foundation for building scalable, maintainable, and reproducible machine learning systems in real-world scenarios.<\/p>\n
\n
Check out\u00a0the\u00a0Full Codes with Notebook here<\/a><\/strong>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0130k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.?\u00a0Connect with us<\/mark><\/a><\/strong><\/p>\n
The post How to Build an End-to-End Production Grade Machine Learning Pipeline with ZenML, Including Custom Materializers, Metadata Tracking, and Hyperparameter Optimization<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we walk thro…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-853","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/853","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=853"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/853\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=853"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=853"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=853"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}