ClearML Tutorial: Open-Source MLOps Platform for Experiment Tracking and Pipeline Automation

ClearML is an open-source MLOps platform that provides a comprehensive solution for experiment tracking, data management, orchestration, and machine learning model deployment. With ClearML, data science teams can manage the entire ML project lifecycle from initial experimentation to production deployment within a single integrated platform.

In this tutorial, we will learn how to use ClearML from installation, basic experiment tracking, dataset management, to pipeline automation. All code examples can be directly practiced in your local environment.

Why ClearML?

Before diving into implementation, here are several reasons why ClearML deserves consideration for your MLOps workflow:

Installation

Installing ClearML SDK

Install the ClearML Python SDK using pip:

pip install clearml

For additional features, you can install with extras:

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

Setting Up ClearML Server

There are two options for ClearML Server:

Register a free account at app.clear.ml, then run:

clearml-init

Enter the credentials (API key, secret, and host) that can be obtained from Settings > Workspace in the ClearML dashboard.

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

After the server is running, access the dashboard at http://localhost:8080 and create new credentials.

Configuration

After running clearml-init, the configuration file will be saved at ~/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="ClearML Tutorial", taskname="First Experiment")


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="ClearML Tutorial", 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 complete!")

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

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


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

disp.plot(ax=ax)
plt.title("Confusion Matrix")
logger.reportmatplotlibfigure("Confusion Matrix", "results", 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("Distribution", "predictions", values=data, xaxis="Value", yaxis="Frequency")

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="ClearML Tutorial", 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("PyTorch training complete!")

from clearml import Task

task = Task.init(projectname="ClearML Tutorial", 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="ClearML Tutorial/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="ClearML Tutorial/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="ClearML Tutorial/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 to {modelpath}")
    return modelpath



@PipelineDecorator.pipeline(
    name="Customer Churn Pipeline",
    project="ClearML Tutorial/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 complete! 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 Pipelines

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 enables task execution on remote machines:

Installing the Agent

pip install clearml-agent
clearml-agent init

Running the Agent

clearml-agent daemon --queue default --foreground

Sending Tasks to a Queue

from clearml import Task

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

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

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

print(f"Task {task.id} sent to queue 'default'")

from clearml import Task

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

clonedtask = Task.clone(sourcetask=originaltask, name="Cloned Experiment")

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="ClearML Tutorial",
    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 provides a model registry for managing model artifacts:

from clearml import OutputModel, InputModel, Task

task = Task.init(projectname="ClearML Tutorial", 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 to: {localpath}")

Best Practices

1. Project Organization

Use a consistent project structure:

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

2. Tagging and Naming Conventions

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

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

3. Environment Reproducibility

ClearML automatically captures requirements, but you can add packages explicitly:

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

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

5. Monitoring and Alerting

from clearml.automation import TaskScheduler

scheduler = TaskScheduler()
scheduler.addtask(

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

6. Use Artifacts for Large Files

Don't log large files as metrics. Use artifacts instead:

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

Conclusion

ClearML is an MLOps platform that offers a comprehensive solution for managing the machine learning lifecycle. Key advantages of ClearML include:

• Easy Adoption: Only 2 lines of code needed to start tracking experiments, without changing your existing workflow.
• Feature-Complete: From experiment tracking, dataset versioning, pipeline automation, to model serving, everything is available in one platform.
• Open-Source: ClearML Server can be self-deployed, giving full control over data and infrastructure.
• Scalable: Supports small teams to enterprise with collaboration features and role-based access control.

By mastering ClearML, you can build MLOps workflows that are reproducible, scalable, and easy to maintain. Start with simple experiment tracking, then gradually adopt dataset management, pipeline automation, and remote execution features as your team's needs grow.

Recommended next steps:

• Explore ClearML Serving for model deployment
• Learn ClearML Sessions for remote development
• Integrate ClearML with CI/CD pipelines using GitHub Actions or GitLab CI