Tutorial ClearML: Platform MLOps Open-Source untuk Experiment Tracking dan Pipeline Automation

ClearML adalah platform MLOps open-source yang menyediakan solusi lengkap untuk experiment tracking, data management, orchestration, dan deployment model machine learning. Dengan ClearML, tim data science dapat mengelola seluruh lifecycle ML project dari eksperimen awal hingga production deployment dalam satu platform terintegrasi.

Dalam tutorial ini, kita akan mempelajari cara menggunakan ClearML mulai dari instalasi, experiment tracking dasar, dataset management, hingga pipeline automation. Semua contoh kode dapat langsung dipraktekkan di environment lokal Anda.

Mengapa ClearML?

Sebelum masuk ke implementasi, ada beberapa alasan mengapa ClearML layak dipertimbangkan untuk workflow MLOps Anda:

Instalasi

Instalasi ClearML SDK

Instal ClearML Python SDK menggunakan pip:

pip install clearml

Untuk fitur tambahan, Anda bisa menginstal dengan extras:

pip install clearml[s3]
pip install clearml[gs]
pip install clearml[azure]

Setup ClearML Server

Ada dua opsi untuk ClearML Server:

Daftar akun gratis di app.clear.ml, lalu jalankan:

clearml-init

Masukkan credentials (API key, secret, dan host) yang bisa didapatkan dari Settings > Workspace di dashboard ClearML.

mkdir -p /opt/clearml && cd /opt/clearml

curl -o docker-compose.yml https://raw.githubusercontent.com/allegroai/clearml-server/master/docker/docker-compose.yml

docker-compose up -d

Setelah server berjalan, akses dashboard di http://localhost:8080 dan buat credentials baru.

Konfigurasi

Setelah menjalankan clearml-init, file konfigurasi akan tersimpan di ~/clearml.conf:

api {
    webserver: https://app.clear.ml

    apiserver: https://api.clear.ml
    filesserver: https://files.clear.ml

    credentials {
        "accesskey" = "YOURACCESSKEY"
        "secretkey" = "YOURSECRETKEY"

    }
}

from clearml import Task

task = Task.init(projectname="Tutorial ClearML", taskname="Eksperimen Pertama")


import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import makeclassification
from sklearn.modelselection import traintestsplit

from sklearn.metrics import accuracyscore, f1score


X, y = makeclassification(
    nsamples=1000,

    nfeatures=20,
    ninformative=10,

    nclasses=2,
    randomstate=42

)
Xtrain, Xtest, ytrain, ytest = traintestsplit(X, y, testsize=0.2, randomstate=42)


params = {
    "nestimators": 100,
    "maxdepth": 10,

    "minsamplessplit": 5,

    "randomstate": 42
}
task.connect(params)

model = RandomForestClassifier(*params)

model.fit(Xtrain, ytrain)

ypred = model.predict(Xtest)
accuracy = accuracyscore(ytest, ypred)

f1 = f1score(ytest, ypred)

logger = task.getlogger()

logger.reportscalar("metrics", "accuracy", value=accuracy, iteration=1)
logger.reportscalar("metrics", "f1score", value=f1, iteration=1)

print(f"Accuracy: {accuracy:.4f}")
print(f"F1 Score: {f1:.4f}")

from clearml import Task
import numpy as np

task = Task.init(projectname="Tutorial ClearML", taskname="Training Loop Metrics")
logger = task.getlogger()


nepochs = 50
for epoch in range(nepochs):

    trainloss = 1.0 / (epoch + 1) + np.random.normal(0, 0.05)
    valloss = 1.2 / (epoch + 1) + np.random.normal(0, 0.08)

    learningrate = 0.001  (0.95  epoch)


    logger.reportscalar("loss", "train", value=trainloss, iteration=epoch)
    logger.reportscalar("loss", "validation", value=valloss, iteration=epoch)
    logger.reportscalar("learningrate", "lr", value=learningrate, iteration=epoch)


print("Training selesai!")

from clearml import Task
import numpy as np
import matplotlib.pyplot as plt
from sklearn.metrics import confusionmatrix, ConfusionMatrixDisplay

task = Task.init(projectname="Tutorial ClearML", taskname="Visualisasi")
logger = task.getlogger()


cm = np.array([[45, 5], [8, 42]])
fig, ax = plt.subplots(figsize=(8, 6))
disp = ConfusionMatrixDisplay(confusionmatrix=cm, displaylabels=["Negatif", "Positif"])

disp.plot(ax=ax)
plt.title("Confusion Matrix")
logger.reportmatplotlibfigure("Confusion Matrix", "hasil", figure=fig)


categories = ["Precision", "Recall", "F1-Score", "Accuracy"]
values = [0.89, 0.84, 0.86, 0.87]
fig2, ax2 = plt.subplots(figsize=(8, 5))
ax2.bar(categories, values, color=["#2196F3", "#4CAF50", "#FF9800", "#9C27B0"])
ax2.setylim(0, 1.0)
ax2.settitle("Model Performance Metrics")

logger.reportmatplotlibfigure("Performance", "barchart", figure=fig2)

data = np.random.randn(1000)
logger.reporthistogram("Distribusi", "prediksi", values=data, xaxis="Nilai", yaxis="Frekuensi")

from clearml import Task
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import numpy as np

task = Task.init(projectname="Tutorial ClearML", taskname="PyTorch Training")

logger = task.getlogger()

params = {
    "batchsize": 32,

    "learningrate": 0.001,
    "epochs": 20,
    "hiddensize": 64

}
task.connect(params)

class SimpleNet(nn.Module):
    def init(self, inputsize, hiddensize, numclasses):
        super().init()
        self.fc1 = nn.Linear(inputsize, hiddensize)
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.3)
        self.fc2 = nn.Linear(hiddensize, hiddensize // 2)
        self.fc3 = nn.Linear(hiddensize // 2, numclasses)

    def forward(self, x):
        x = self.dropout(self.relu(self.fc1(x)))
        x = self.relu(self.fc2(x))
        return self.fc3(x)

X = np.random.randn(1000, 20).astype(np.float32)
y = (X[:, 0] + X[:, 1] > 0).astype(np.int64)

dataset = TensorDataset(torch.fromnumpy(X), torch.fromnumpy(y))
trainloader = DataLoader(dataset, batchsize=params["batchsize"], shuffle=True)


model = SimpleNet(20, params["hiddensize"], 2)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=params["learningrate"])


for epoch in range(params["epochs"]):
    model.train()
    totalloss = 0
    correct = 0
    total = 0

    for batchX, batchy in trainloader:

        optimizer.zerograd()
        outputs = model(batchX)

        loss = criterion(outputs, batchy)
        loss.backward()
        optimizer.step()

        totalloss += loss.item()

        , predicted = torch.max(outputs, 1)
        total += batchy.size(0)

        correct += (predicted == batchy).sum().item()

    avgloss = totalloss / len(trainloader)

    accuracy = correct / total

    logger.reportscalar("training", "loss", value=avgloss, iteration=epoch)

    logger.reportscalar("training", "accuracy", value=accuracy, iteration=epoch)

torch.save(model.statedict(), "model.pth")

task.uploadartifact("modelweights", artifactobject="model.pth")

print("Training PyTorch selesai!")

from clearml import Task

task = Task.init(projectname="Tutorial ClearML", taskname="HuggingFace Fine-tuning")

from transformers import (
    AutoTokenizer,
    AutoModelForSequenceClassification,
    TrainingArguments,
    Trainer
)
from datasets import loaddataset

import numpy as np
from sklearn.metrics import accuracyscore

dataset = loaddataset("imdb", split={"train": "train[:1000]", "test": "test[:200]"})

tokenizer = AutoTokenizer.frompretrained("distilbert-base-uncased")

def tokenizefunction(examples):

    return tokenizer(examples["text"], padding="maxlength", truncation=True, maxlength=256)


tokenized = dataset.map(tokenizefunction, batched=True)

model = AutoModelForSequenceClassification.frompretrained(

    "distilbert-base-uncased", numlabels=2
)

def computemetrics(evalpred):
    logits, labels = evalpred

    predictions = np.argmax(logits, axis=-1)
    return {"accuracy": accuracyscore(labels, predictions)}

trainingargs = TrainingArguments(

    outputdir="./results",
    numtrainepochs=3,
    perdevicetrainbatchsize=16,
    perdeviceevalbatchsize=16,
    evaluationstrategy="epoch",

    savestrategy="epoch",
    loggingsteps=10,

    loadbestmodelatend=True,

)

trainer = Trainer(
    model=model,
    args=trainingargs,
    traindataset=tokenized["train"],

    evaldataset=tokenized["test"],
    computemetrics=computemetrics,
)

trainer.train()

from clearml import Dataset

dataset = Dataset.create(
    datasetname="Customer Churn Dataset",

    datasetproject="Tutorial ClearML/Datasets"
)

dataset.addfiles(path="./data/customerchurn.csv")
dataset.addfiles(path="./data/features/")


dataset.getlogger().reporttable(

    "Dataset Info",
    "summary",
    tableplot=[
        ["Total Records", "10,000"],
        ["Features", "15"],
        ["Target", "churn (binary)"],
        ["Missing Values", "2.3%"]
    ]
)

dataset.upload()
dataset.finalize()

print(f"Dataset ID: {dataset.id}")

from clearml import Dataset

datasetpath = Dataset.get(

    datasetname="Customer Churn Dataset",
    datasetproject="Tutorial ClearML/Datasets"

).getlocalcopy()


import pandas as pd
df = pd.readcsv(f"{datasetpath}/customerchurn.csv")
print(f"Dataset loaded: {df.shape}")

from clearml import Dataset

childdataset = Dataset.create(

    datasetname="Customer Churn Dataset",
    datasetproject="Tutorial ClearML/Datasets",

    parentdatasets=[parentdatasetid]
)

childdataset.addfiles(path="./data/newrecords.csv")

childdataset.removefiles(match="outdatedrecords.csv")

childdataset.upload()

childdataset.finalize()

from clearml import PipelineDecorator

@PipelineDecorator.component(
    returnvalues=["Xtrain", "Xtest", "ytrain", "ytest"],

    cache=True
)
def stepdatapreparation(datasetname, testsize=0.2):

    import pandas as pd
    from sklearn.modelselection import traintestsplit
    from clearml import Dataset

    datasetpath = Dataset.get(datasetname=datasetname).getlocalcopy()

    df = pd.readcsv(f"{datasetpath}/customerchurn.csv")

    X = df.drop("churn", axis=1)
    y = df["churn"]

    Xtrain, Xtest, ytrain, ytest = traintestsplit(
        X, y, testsize=testsize, randomstate=42, stratify=y

    )

    print(f"Train size: {len(Xtrain)}, Test size: {len(Xtest)}")

    return Xtrain, Xtest, ytrain, ytest



@PipelineDecorator.component(
    returnvalues=["Xtrainscaled", "Xtestscaled", "scaler"],
    cache=True
)
def stepfeatureengineering(Xtrain, Xtest):
    from sklearn.preprocessing import StandardScaler
    import numpy as np

    scaler = StandardScaler()
    Xtrainscaled = scaler.fittransform(Xtrain)
    Xtestscaled = scaler.transform(Xtest)


    print(f"Features scaled. Mean: {np.mean(Xtrainscaled):.4f}")

    return Xtrainscaled, Xtestscaled, scaler



@PipelineDecorator.component(
    returnvalues=["model", "metrics"],
    cache=False
)
def stepmodeltraining(Xtrain, ytrain, Xtest, ytest, modelparams):

    from sklearn.ensemble import GradientBoostingClassifier
    from sklearn.metrics import accuracyscore, precisionscore, recallscore, f1score


    model = GradientBoostingClassifier(modelparams)
    model.fit(Xtrain, ytrain)

    ypred = model.predict(Xtest)

    metrics = {
        "accuracy": accuracyscore(ytest, ypred),

        "precision": precisionscore(ytest, ypred),
        "recall": recallscore(ytest, ypred),

        "f1score": f1score(ytest, ypred)

    }

    print(f"Model trained. Accuracy: {metrics['accuracy']:.4f}")
    return model, metrics


@PipelineDecorator.component(returnvalues=["modelpath"])

def stepmodelexport(model, scaler, metrics):

    import joblib
    from clearml import Task

    modelpath = "modelproduction.joblib"

    joblib.dump({"model": model, "scaler": scaler, "metrics": metrics}, modelpath)

    task = Task.currenttask()

    task.uploadartifact("productionmodel", artifactobject=modelpath)


    print(f"Model exported ke {modelpath}")
    return modelpath



@PipelineDecorator.pipeline(
    name="Customer Churn Pipeline",
    project="Tutorial ClearML/Pipelines",
    version="1.0"
)
def churnpredictionpipeline(datasetname, modelparams):

    Xtrain, Xtest, ytrain, ytest = stepdatapreparation(

        datasetname=datasetname

    )

    Xtrainscaled, Xtestscaled, scaler = stepfeatureengineering(

        Xtrain=Xtrain, Xtest=Xtest

    )

    model, metrics = stepmodeltraining(

        Xtrain=Xtrainscaled,
        ytrain=ytrain,
        Xtest=Xtestscaled,

        ytest=ytest,

        modelparams=modelparams

    )

    modelpath = stepmodelexport(
        model=model, scaler=scaler, metrics=metrics
    )

    print(f"Pipeline selesai! Metrics: {metrics}")
    return modelpath



if name == "main":
    PipelineDecorator.runlocally()

    modelparams = {

        "nestimators": 200,
        "maxdepth": 5,

        "learningrate": 0.1,
        "subsample": 0.8
    }

    churnpredictionpipeline(
        datasetname="Customer Churn Dataset",

        modelparams=modelparams

    )

if name == "main":
    churnpredictionpipeline(

        datasetname="Customer Churn Dataset",
        modelparams=modelparams
    )

Scheduling Pipeline

from clearml.automation import TriggerScheduler

scheduler = TriggerScheduler()

scheduler.addtask(

    schedulefunction=lambda: churnpredictionpipeline(
        datasetname="Customer Churn Dataset",

        modelparams={"nestimators": 200, "maxdepth": 5}
    ),
    name="Daily Churn Pipeline",
    minute=0,
    hour=6
)

scheduler.start()

ClearML Agent: Remote Execution

ClearML Agent memungkinkan eksekusi task di mesin remote:

Instalasi Agent

pip install clearml-agent
clearml-agent init

Menjalankan Agent

clearml-agent daemon --queue default --foreground

Mengirim Task ke Queue

from clearml import Task

task = Task.init(projectname="Tutorial ClearML", taskname="Remote Training")

params = {"epochs": 50, "batchsize": 64, "learningrate": 0.001}
task.connect(params)

Task.enqueue(task, queuename="default")

print(f"Task {task.id} dikirim ke queue 'default'")

from clearml import Task

originaltask = Task.gettask(taskid="abc123")

clonedtask = Task.clone(sourcetask=originaltask, name="Eksperimen Clone")

clonedparams = clonedtask.getparameters()
clonedparams["General/learningrate"] = 0.0005
clonedtask.setparameters(clonedparams)


Task.enqueue(clonedtask, queuename="gpu-queue")

from clearml import Task
from clearml.automation import (
    UniformParameterRange,
    UniformIntegerParameterRange,
    DiscreteParameterRange,
    HyperParameterOptimizer,
    RandomSearch
)

task = Task.init(
    projectname="Tutorial ClearML",
    taskname="HPO Controller",

    tasktype=Task.TaskTypes.optimizer
)

optimizer = HyperParameterOptimizer(
    basetaskid="BASETASKID",
    hyperparameters=[

        UniformParameterRange("General/learningrate", minvalue=0.0001, maxvalue=0.01),
        UniformIntegerParameterRange("General/batchsize", minvalue=16, maxvalue=128),

        UniformIntegerParameterRange("General/epochs", minvalue=10, maxvalue=100),

        DiscreteParameterRange("General/optimizer", values=["adam", "sgd", "adamw"]),
    ],
    objectivemetrictitle="metrics",

    objectivemetricseries="accuracy",

    objectivemetricsign="max",

    optimizerclass=RandomSearch,
    maxnumberofconcurrenttasks=4,
    totalmaxjobs=20,
    miniterationperjob=0,

    maxiterationperjob=150,
)

optimizer.setreportperiod(1)
optimizer.start()
optimizer.wait()

best = optimizer.gettopexperiments(topk=3)

print(f"Best experiment: {best[0].id}")
print(f"Best accuracy: {best[0].getlastscalarmetrics()}")

optimizer.stop()

Model Registry

ClearML menyediakan model registry untuk mengelola model artifacts:

from clearml import OutputModel, InputModel, Task

task = Task.init(projectname="Tutorial ClearML", taskname="Model Registry")

outputmodel = OutputModel(

    task=task,
    name="Churn Prediction Model v1",
    tags=["production", "churn", "gradient-boosting"],
    framework="scikit-learn"
)

outputmodel.updateweights(weightsfilename="modelproduction.joblib")


outputmodel.setmetadata({

    "accuracy": "0.92",
    "f1score": "0.89",
    "trainingdataversion": "v2.1",
    "featurecount": "15"

})

model = InputModel(modelid="MODELID")

localpath = model.getweights()

print(f"Model downloaded ke: {localpath}")

Best Practices

1. Organisasi Project

Gunakan struktur project yang konsisten:

Company/
├── Team-A/
│   ├── Experiments/
│   ├── Pipelines/
│   └── Datasets/
├── Team-B/
│   ├── Experiments/
│   └── Pipelines/
└── Production/
    ├── Models/
    └── Monitoring/

2. Tagging dan Naming Convention

task = Task.init(
    projectname="Fraud Detection/Experiments",

    taskname="XGBoost v2 - balanced sampling",
    tags=["xgboost", "balanced", "v2", "candidate"]
)

3. Environment Reproducibility

ClearML secara otomatis menangkap requirements, tetapi Anda bisa menambahkan packages secara eksplisit:

task.setpackages(["torch==2.0.1", "transformers==4.30.0"])

@PipelineDecorator.component(cache=True)
def expensivepreprocessing(data):
    ...

5. Monitoring dan Alerting

from clearml.automation import TaskScheduler

scheduler = TaskScheduler()
scheduler.addtask(

    taskid="monitoringtaskid",
    queue="monitoring",
    minute=0,
    hour="*/4"
)

6. Gunakan Artifacts untuk Large Files

Jangan log file besar sebagai metrics. Gunakan artifacts:

task.uploadartifact("largedataset", artifactobject=dataframe)
task.uploadartifact("modelcheckpoint", artifactobject="checkpoint.pt")

Kesimpulan

ClearML adalah platform MLOps yang menawarkan solusi menyeluruh untuk mengelola lifecycle machine learning. Beberapa keunggulan utama ClearML:

• Mudah Diadopsi: Hanya perlu 2 baris kode untuk mulai tracking eksperimen, tanpa perlu mengubah workflow yang sudah ada.
• Fitur Lengkap: Dari experiment tracking, dataset versioning, pipeline automation, hingga model serving, semua tersedia dalam satu platform.
• Open-Source: ClearML Server dapat di-deploy sendiri, memberikan kontrol penuh atas data dan infrastruktur.
• Skalabel: Mendukung tim kecil hingga enterprise dengan fitur collaboration dan role-based access control.

Dengan menguasai ClearML, Anda dapat membangun workflow MLOps yang reproducible, scalable, dan mudah di-maintain. Mulai dari experiment tracking sederhana, lalu secara bertahap adopsi fitur dataset management, pipeline automation, dan remote execution sesuai kebutuhan tim Anda.

Langkah selanjutnya yang disarankan:

• Eksplorasi ClearML Serving untuk model deployment
• Pelajari ClearML Sessions untuk remote development
• Integrasikan ClearML dengan CI/CD pipeline menggunakan GitHub Actions atau GitLab CI