ZenML: Framework Pipeline MLOps yang Modular dan Cloud-Agnostic

Pendahuluan

Membangun model machine learning yang akurat hanyalah sebagian kecil dari tantangan dalam produksi ML. Tantangan sesungguhnya terletak pada bagaimana membuat pipeline yang reproducible, scalable, dan mudah di-maintain. ZenML hadir sebagai framework MLOps open-source yang dirancang untuk mengatasi masalah ini dengan pendekatan yang modular dan cloud-agnostic.

ZenML memungkinkan data scientist dan ML engineer untuk mendefinisikan pipeline ML menggunakan Python decorator yang sederhana, sambil tetap memberikan fleksibilitas penuh untuk mengintegrasikan berbagai tool dan platform cloud. Dengan konsep "stacks" yang unik, Anda bisa berpindah dari local development ke production di cloud tanpa mengubah kode pipeline.

Dalam tutorial ini, kita akan mempelajari ZenML dari dasar hingga membangun pipeline end-to-end yang mencakup data ingestion, preprocessing, training, evaluation, dan deployment.

Prasyarat

Sebelum memulai, pastikan Anda memiliki:

• Python 3.8 atau lebih baru
• pip package manager
• Pemahaman dasar tentang machine learning
• Familiar dengan scikit-learn atau framework ML lainnya

Instalasi ZenML

Instalasi ZenML sangat mudah menggunakan pip:

pip install zenml

Untuk instalasi dengan integrasi tambahan:

pip install "zenml[server]"

Setelah instalasi, inisialisasi ZenML repository di project Anda:

zenml init

Jalankan ZenML dashboard secara lokal:

zenml login --local

Dashboard akan tersedia di http://127.0.0.1:8237 dan memberikan visualisasi lengkap pipeline, artifacts, dan stacks Anda.

Konsep Dasar: Steps dan Pipelines

Mendefinisikan Steps dengan @step

Step adalah unit kerja terkecil dalam ZenML. Setiap step adalah sebuah fungsi Python yang di-dekorasi dengan @step:

from zenml import step
import pandas as pd

@step
def loaddata() -> pd.DataFrame:

    """Memuat dataset dari sumber data."""
    df = pd.readcsv("data/trainingdata.csv")

    return df

@step
def preprocessdata(df: pd.DataFrame) -> pd.DataFrame:
    """Membersihkan dan memproses data."""
    df = df.dropna()
    df = df.dropduplicates()

    return df

from zenml import pipeline

@pipeline
def trainingpipeline():
    """Pipeline untuk training model ML."""
    data = loaddata()

    processeddata = preprocessdata(data)

    model = trainmodel(processeddata)

    metrics = evaluatemodel(model, processeddata)

    return metrics

if name == "main":
    trainingpipeline()

Parameterisasi Steps

Anda bisa membuat steps yang menerima parameter konfigurasi:

from zenml import step
from pydantic import BaseModel

class TrainingConfig(BaseModel):
    learningrate: float = 0.01

    nestimators: int = 100
    maxdepth: int = 5


@step
def trainmodel(
    data: pd.DataFrame,
    config: TrainingConfig = TrainingConfig()
) -> object:
    """Melatih model dengan konfigurasi yang dapat disesuaikan."""
    from sklearn.ensemble import RandomForestClassifier

    X = data.drop("target", axis=1)
    y = data["target"]

    model = RandomForestClassifier(
        nestimators=config.nestimators,
        maxdepth=config.maxdepth
    )
    model.fit(X, y)
    return model

Artifacts dan Materializers

Memahami Artifacts

Setiap output dari step secara otomatis disimpan sebagai artifact. Artifact adalah data yang dihasilkan dan dikonsumsi oleh steps dalam pipeline. ZenML melacak setiap artifact dengan metadata lengkap termasuk versi, tipe, dan lineage.

from zenml import step, logartifactmetadata

@step
def trainmodel(data: pd.DataFrame) -> object:

    model = RandomForestClassifier()
    model.fit(Xtrain, ytrain)


    logartifactmetadata(

        artifactname="model",
        metadata={
            "accuracy": float(model.score(Xtest, ytest)),
            "nfeatures": Xtrain.shape[1],
            "algorithm": "RandomForest"
        }
    )
    return model

Custom Materializers

Materializer mengontrol bagaimana artifact diserialisasi dan dideserialisasi. Anda bisa membuat custom materializer untuk tipe data khusus:

from zenml.materializers import BaseMaterializer
from zenml.enums import ArtifactType
import json
import os

class CustomModelMaterializer(BaseMaterializer):
    ASSOCIATEDTYPES = (CustomModel,)

    ASSOCIATEDARTIFACTTYPE = ArtifactType.MODEL


    def load(self, datatype):
        filepath = os.path.join(self.uri, "model.json")
        with open(filepath, "r") as f:
            data = json.load(f)
        return CustomModel.fromdict(data)


    def save(self, model):
        filepath = os.path.join(self.uri, "model.json")
        with open(filepath, "w") as f:
            json.dump(model.todict(), f)

Stacks: Infrastruktur yang Modular

Konsep Stack

Stack adalah konfigurasi infrastruktur yang mendefinisikan di mana dan bagaimana pipeline dijalankan. Setiap stack terdiri dari beberapa komponen:

• Orchestrator: Menjalankan pipeline (local, Airflow, Kubeflow, dll.)
• Artifact Store: Menyimpan artifacts (local, S3, GCS, Azure Blob)
• Experiment Tracker: Melacak eksperimen (MLflow, WandB, Neptune)
• Model Deployer: Men-deploy model (MLflow, Seldon, BentoML)
• Container Registry: Menyimpan Docker images
• Step Operator: Menjalankan steps di environment khusus (SageMaker, Vertex AI)

Mengelola Stacks

# Melihat stack yang tersedia
zenml stack list

Mendaftarkan komponen baru
zenml artifact-store register mys3store \
    --flavor=s3 \
    --path=s3://my-bucket/zenml

zenml orchestrator register mykubeflow \

    --flavor=kubeflow \
    --kubernetescontext=my-cluster

Membuat stack baru
zenml stack register productionstack \

    --orchestrator=mykubeflow \
    --artifact-store=mys3store

Mengaktifkan stack
zenml stack set productionstack

# Registrasi experiment tracker MLflow
zenml experiment-tracker register mlflowtracker \
    --flavor=mlflow \
    --trackinguri=http://mlflow-server:5000


Registrasi model deployer
zenml model-deployer register mlflowdeployer \
    --flavor=mlflow

Stack lengkap untuk production
zenml stack register fullstack \

    --orchestrator=mykubeflow \
    --artifact-store=mys3store \
    --experiment-tracker=mlflowtracker \

    --model-deployer=mlflowdeployer

Integrasi dengan MLflow dan Weights & Biases

Integrasi MLflow

Instal integrasi MLflow:

zenml integration install mlflow -y

Gunakan MLflow experiment tracker dalam step:

from zenml import step
from zenml.integrations.mlflow.flavors.mlflowexperimenttrackerflavor import (

    MLFlowExperimentTrackerSettings
)
import mlflow

mlflowsettings = MLFlowExperimentTrackerSettings(
    experimentname="myexperiment"
)

@step(
    experimenttracker="mlflowtracker",
    settings={"experimenttracker": mlflowsettings}
)
def trainwithmlflow(data: pd.DataFrame) -> object:
    mlflow.autolog()

    model = RandomForestClassifier(nestimators=100)

    model.fit(Xtrain, ytrain)


    accuracy = model.score(Xtest, ytest)

    mlflow.logmetric("accuracy", accuracy)
    mlflow.logparam("nestimators", 100)

    return model

Integrasi Weights & Biases

zenml integration install wandb -y

from zenml.integrations.wandb.flavors.wandbexperimenttrackerflavor import (
    WandbExperimentTrackerSettings
)
import wandb

wandbsettings = WandbExperimentTrackerSettings(
    project="my-zenml-project",
    entity="my-team"
)

@step(
    experimenttracker="wandbtracker",
    settings={"experimenttracker": wandbsettings}
)
def trainwithwandb(data: pd.DataFrame) -> object:
    config = {"nestimators": 100, "maxdepth": 5}
    wandb.config.update(config)

    model = RandomForestClassifier(*config)

    model.fit(Xtrain, ytrain)

    wandb.log({"accuracy": model.score(Xtest, ytest)})
    return model

from zenml import step, pipeline
from zenml.integrations.mlflow.steps import mlflowmodeldeployerstep


@step
def deploymenttrigger(accuracy: float) -> bool:
    """Menentukan apakah model layak di-deploy."""
    return accuracy > 0.85

@pipeline
def deploymentpipeline():

    data = loaddata()
    processed = preprocessdata(data)

    model = trainmodel(processed)
    accuracy = evaluatemodel(model, processed)

    shoulddeploy = deploymenttrigger(accuracy)


    if shoulddeploy:
        mlflowmodeldeployerstep(

            model=model,
            deploydecision=shoulddeploy,

            workers=3
        )

# Menonaktifkan caching untuk step tertentu
@step(enablecache=False)
def alwaysfreshdata() -> pd.DataFrame:
    """Step ini selalu dijalankan ulang."""
    return pd.readcsv("livedata.csv")

Menonaktifkan caching untuk seluruh pipeline
@pipeline(enablecache=False)

def nocachepipeline():

    data = alwaysfreshdata()

    process(data)

from zenml.config.schedule import Schedule

Jadwalkan pipeline untuk berjalan setiap hari
schedule = Schedule(
    cronexpression="0 8   "  # Setiap hari jam 8 pagi
)

trainingpipeline = trainingpipeline.withoptions(

    schedule=schedule
)
trainingpipeline()

Anda juga bisa menjadwalkan dengan interval:

from datetime import timedelta

schedule = Schedule(
    intervalsecond=timedelta(hours=6),

    starttime="2026-01-01T00:00:00"
)

ZenML Dashboard

ZenML menyediakan dashboard web yang powerful untuk monitoring:

# Login ke ZenML server
zenml login --local

Atau koneksi ke remote server
zenml login https://my-zenml-server.com

Dashboard menyediakan:

• Pipeline Runs: Visualisasi DAG pipeline dan status setiap step
• Artifacts: Browser untuk melihat semua artifact yang dihasilkan
• Stacks: Manajemen stack dan komponen infrastruktur
• Model Registry: Tracking model versions dan deployment status

Contoh Praktis: Pipeline ML End-to-End

Berikut contoh lengkap pipeline dari data ingestion hingga deployment:

import pandas as pd
import numpy as np
from sklearn.modelselection import traintestsplit

from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracyscore, classificationreport

from zenml import step, pipeline, logartifactmetadata, ArtifactConfig

from zenml.client import Client
from typing import Tuple
from typingextensions import Annotated
from pydantic import BaseModel

Konfigurasi
class ModelConfig(BaseModel):
    testsize: float = 0.2

    nestimators: int = 200
    maxdepth: int = 10

    randomstate: int = 42
    minaccuracy: float = 0.85


Step 1: Data Ingestion
@step
def ingestdata() -> Annotated[pd.DataFrame, "rawdata"]:

    """Mengambil data dari sumber."""
    from sklearn.datasets import loadiris
    iris = loadiris(asframe=True)
    df = iris.frame
    df.columns = ["sepallength", "sepalwidth",
                   "petallength", "petalwidth", "target"]

    logartifactmetadata(
        artifactname="rawdata",
        metadata={
            "numrows": len(df),

            "numcolumns": len(df.columns),
            "columns": list(df.columns)
        }
    )
    return df

Step 2: Data Preprocessing
@step
def preprocess(
    df: pd.DataFrame,
    config: ModelConfig = ModelConfig()
) -> Tuple[
    Annotated[np.ndarray, "Xtrain"],

    Annotated[np.ndarray, "Xtest"],
    Annotated[np.ndarray, "ytrain"],

    Annotated[np.ndarray, "ytest"],
    Annotated[StandardScaler, "scaler"]
]:
    """Preprocessing: split, scale, encode."""
    X = df.drop("target", axis=1).values
    y = df["target"].values

    Xtrain, Xtest, ytrain, ytest = traintestsplit(
        X, y,
        testsize=config.testsize,
        randomstate=config.randomstate,
        stratify=y
    )

    scaler = StandardScaler()
    Xtrain = scaler.fittransform(Xtrain)

    Xtest = scaler.transform(Xtest)


    return Xtrain, Xtest, ytrain, ytest, scaler


Step 3: Model Training
@step
def train(
    Xtrain: np.ndarray,
    ytrain: np.ndarray,

    config: ModelConfig = ModelConfig()
) -> Annotated[RandomForestClassifier, "trainedmodel"]:
    """Melatih model RandomForest."""
    model = RandomForestClassifier(
        nestimators=config.nestimators,
        maxdepth=config.maxdepth,
        randomstate=config.randomstate,
        njobs=-1

    )
    model.fit(Xtrain, ytrain)


    logartifactmetadata(

        artifactname="trainedmodel",

        metadata={
            "algorithm": "RandomForestClassifier",
            "nestimators": config.nestimators,

            "maxdepth": config.maxdepth

        }
    )
    return model

Step 4: Model Evaluation
@step
def evaluate(
    model: RandomForestClassifier,
    Xtest: np.ndarray,
    ytest: np.ndarray,

    config: ModelConfig = ModelConfig()
) -> Annotated[float, "accuracy"]:
    """Mengevaluasi performa model."""
    ypred = model.predict(Xtest)

    accuracy = accuracyscore(ytest, ypred)
    report = classificationreport(ytest, ypred, outputdict=True)

    logartifactmetadata(
        artifactname="accuracy",

        metadata={
            "accuracy": accuracy,
            "precisionmacro": report["macro avg"]["precision"],
            "recallmacro": report["macro avg"]["recall"],

            "f1macro": report["macro avg"]["f1-score"]
        }
    )

    print(f"Model Accuracy: {accuracy:.4f}")
    print(classificationreport(ytest, ypred))

    return accuracy

Step 5: Deployment Decision
@step
def decidedeployment(
    accuracy: float,
    config: ModelConfig = ModelConfig()
) -> Annotated[bool, "deploydecision"]:

    """Memutuskan apakah model layak di-deploy."""
    shoulddeploy = accuracy >= config.minaccuracy

    if shoulddeploy:
        print(f"Model LAYAK deploy (accuracy: {accuracy:.4f} >= {config.minaccuracy})")

    else:
        print(f"Model TIDAK layak deploy (accuracy: {accuracy:.4f} < {config.minaccuracy})")
    return shoulddeploy


Pipeline Definition
@pipeline
def mlpipeline():
    """Pipeline ML end-to-end."""
    rawdata = ingestdata()
    Xtrain, Xtest, ytrain, ytest, scaler = preprocess(rawdata)

    model = train(Xtrain, ytrain)

    accuracy = evaluate(model, Xtest, ytest)

    deploydecision = decidedeployment(accuracy)

    return deploydecision

Menjalankan pipeline
if name == "main":
    run = mlpipeline()

    print(f"Pipeline selesai! Run ID: {run.id}")

from zenml.config import DockerSettings

dockersettings = DockerSettings(
    requirements=["scikit-learn==1.3.0", "pandas==2.0.0"],
    requiredintegrations=["mlflow"]

)

@pipeline(settings={"docker": dockersettings})
def reproducible_pipeline():
    ...

Kesimpulan

ZenML menyediakan framework yang elegan untuk membangun pipeline MLOps yang reproducible dan modular. Dengan konsep stacks yang fleksibel, Anda bisa dengan mudah beralih antara development lokal dan production di cloud tanpa mengubah kode pipeline. Integrasi yang luas dengan tools populer seperti MLflow dan Weights & Biases membuatnya menjadi pilihan yang solid untuk tim yang ingin menerapkan MLOps best practices.

Fitur-fitur seperti caching otomatis, artifact tracking, dan pipeline scheduling menjadikan ZenML solusi yang komprehensif untuk mengelola lifecycle machine learning dari eksperimen hingga production.