Tutorial Lengkap Evidently AI: ML Model Monitoring dan Data Quality

Evidently adalah library Python open-source untuk mengevaluasi, testing, dan monitoring model machine learning di production. Library ini membantu mendeteksi data drift, degradasi model, dan masalah kualitas data sebelum berdampak pada bisnis Anda.

Mengapa Evidently?

Keunggulan Evidently:

• Data drift detection: Monitor perubahan input data
• Model performance tracking: Track metrics seiring waktu
• Visual reports: Dashboard HTML interaktif
• Test suites: Automated quality checks
• Easy integration: Works dengan ML framework apapun

Use Cases:

• Production model monitoring
• Data quality validation
• A/B testing analysis
• Pre-deployment validation
• Debugging masalah model

Instalasi

# Basic installation
pip install evidently

Dengan visualization support
pip install evidently[notebooks]

Verify installation
python -c "import evidently; print(evidently.version)"

Quick Start

1. Basic Data Drift Report

import pandas as pd
from evidently.report import Report
from evidently.metricpreset import DataDriftPreset


Load reference (training) dan current (production) data
referencedata = pd.readcsv("trainingdata.csv")
currentdata = pd.readcsv("productiondata.csv")


Create report
report = Report(metrics=[DataDriftPreset()])

Run analysis
report.run(
    referencedata=referencedata,

    currentdata=currentdata

)

Save sebagai HTML
report.savehtml("driftreport.html")


Get results sebagai dict
results = report.asdict()
print(f"Dataset drift detected: {results['metrics'][0]['result']['datasetdrift']}")

2. Model Performance Report

from evidently.report import Report
from evidently.metricpreset import ClassificationPreset

Data dengan predictions dan labels
data = pd.DataFrame({
    "feature1": [1.0, 2.0, 3.0, 4.0, 5.0],
    "feature2": [0.5, 1.5, 2.5, 3.5, 4.5],
    "prediction": [0, 1, 1, 0, 1],
    "target": [0, 1, 0, 0, 1],
})

Classification report
report = Report(metrics=[ClassificationPreset()])
report.run(currentdata=data, columnmapping={
    "target": "target",
    "prediction": "prediction"
})
report.savehtml("classificationreport.html")

Metric Presets

1. Data Drift Preset

from evidently.report import Report
from evidently.metricpreset import DataDriftPreset


report = Report(metrics=[
    DataDriftPreset(
        columns=["feature1", "feature2", "feature3"],  # Specific columns
        driftshare=0.5,  # Threshold untuk dataset drift
    )
])

report.run(referencedata=refdf, currentdata=currdf)

2. Data Quality Preset

from evidently.metricpreset import DataQualityPreset

report = Report(metrics=[DataQualityPreset()])
report.run(currentdata=data)

Check untuk:
- Missing values
- Duplicates
- Constant columns
- Empty columns
- New/missing categories

3. Target Drift Preset

from evidently.metricpreset import TargetDriftPreset

report = Report(metrics=[TargetDriftPreset()])
report.run(
    referencedata=refdf,

    currentdata=currdf,

    columnmapping={"target": "label"}
)

4. Classification Preset

from evidently.metricpreset import ClassificationPreset

report = Report(metrics=[ClassificationPreset()])
report.run(
    currentdata=data,
    columnmapping={

        "target": "actual",
        "prediction": "predicted",
        "poslabel": 1  # Untuk binary classification
    }
)

5. Regression Preset

from evidently.metricpreset import RegressionPreset

report = Report(metrics=[RegressionPreset()])
report.run(
    currentdata=data,
    columnmapping={

        "target": "actualprice",
        "prediction": "predictedprice"

    }
)

Individual Metrics

1. Data Drift Metrics

from evidently.metrics import (
    DatasetDriftMetric,
    ColumnDriftMetric,
    DataDriftTable,
)

report = Report(metrics=[
    DatasetDriftMetric(),
    ColumnDriftMetric(columnname="feature1"),
    ColumnDriftMetric(columnname="feature2"),

    DataDriftTable(),
])

report.run(referencedata=refdf, currentdata=currdf)

2. Data Quality Metrics

from evidently.metrics import (
    DatasetSummaryMetric,
    ColumnSummaryMetric,
    DatasetMissingValuesMetric,
    DatasetCorrelationsMetric,
)

report = Report(metrics=[
    DatasetSummaryMetric(),
    ColumnSummaryMetric(columnname="feature1"),
    DatasetMissingValuesMetric(),
    DatasetCorrelationsMetric(),
])

report.run(currentdata=data)

3. Classification Metrics

from evidently.metrics import (
    ClassificationQualityMetric,
    ClassificationConfusionMatrix,
    ClassificationQualityByClass,
)

report = Report(metrics=[
    ClassificationQualityMetric(),
    ClassificationConfusionMatrix(),
    ClassificationQualityByClass(),
])

report.run(currentdata=data, columnmapping=columnmapping)

4. Regression Metrics

from evidently.metrics import (
    RegressionQualityMetric,
    RegressionPredictedVsActual,
    RegressionErrorDistribution,
)

report = Report(metrics=[
    RegressionQualityMetric(),
    RegressionPredictedVsActual(),
    RegressionErrorDistribution(),
])

report.run(currentdata=data, columnmapping=columnmapping)

Test Suites

1. Basic Test Suite

from evidently.testsuite import TestSuite
from evidently.testpreset import DataDriftTestPreset


Create test suite
suite = TestSuite(tests=[DataDriftTestPreset()])

Run tests
suite.run(referencedata=refdf, currentdata=currdf)


Check results
if suite.asdict()["summary"]["allpassed"]:

    print("Semua tests passed!")
else:
    print("Beberapa tests failed!")

Save report
suite.savehtml("testresults.html")

2. Test Presets

from evidently.testpreset import (
    DataDriftTestPreset,
    DataQualityTestPreset,
    DataStabilityTestPreset,
    NoTargetPerformanceTestPreset,
    RegressionTestPreset,
    BinaryClassificationTestPreset,
    MulticlassClassificationTestPreset,
)

Data drift tests
suite = TestSuite(tests=[DataDriftTestPreset()])

Data quality tests
suite = TestSuite(tests=[DataQualityTestPreset()])

Classification tests
suite = TestSuite(tests=[BinaryClassificationTestPreset()])

Regression tests
suite = TestSuite(tests=[RegressionTestPreset()])

3. Individual Tests

from evidently.tests import (
    TestNumberOfRows,
    TestNumberOfColumns,
    TestColumnsType,
    TestColumnDrift,
    TestShareOfMissingValues,
    TestMeanInNSigmas,
    TestAccuracyScore,
    TestPrecisionScore,
    TestRecallScore,
    TestF1Score,
)

suite = TestSuite(tests=[
    # Data integrity
    TestNumberOfRows(gte=1000),
    TestNumberOfColumns(eq=10),

    # Missing values
    TestShareOfMissingValues(lt=0.05),

    # Feature drift
    TestColumnDrift(columnname="feature1"),


    # Statistical tests
    TestMeanInNSigmas(columnname="feature1", n=3),

    # Model performance
    TestAccuracyScore(gte=0.85),
    TestPrecisionScore(gte=0.80),
    TestRecallScore(gte=0.80),
    TestF1Score(gte=0.80),
])

suite.run(
    referencedata=refdf,
    currentdata=currdf,
    columnmapping=columnmapping
)

4. Custom Conditions

from evidently.tests import TestColumnValueMean, TestColumnValueMin

suite = TestSuite(tests=[
    TestColumnValueMean(
        columnname="price",

        gte=100,
        lte=1000,
    ),
    TestColumnValueMin(
        columnname="quantity",
        gte=0,  # Tidak ada quantity negatif
    ),
])

Column Mapping

from evidently import ColumnMapping

Define column mapping
columnmapping = ColumnMapping(

    target="actuallabel",
    prediction="predictedlabel",

    numericalfeatures=["feature1", "feature2", "feature3"],
    categoricalfeatures=["category1", "category2"],

    datetime="timestamp",
    id="userid",
    task="classification",  # atau "regression"
)

Gunakan di report
report.run(
    currentdata=data,

    columnmapping=columnmapping

)

Monitoring Dashboard

1. Create Monitoring Project

from evidently.ui.workspace import Workspace

Create workspace
ws = Workspace.create("myworkspace")

Create project
project = ws.createproject("My ML Model")

project.description = "Production monitoring untuk recommendation model"
project.save()

2. Tambahkan Reports ke Project

from evidently.report import Report
from evidently.metricpreset import DataDriftPreset, ClassificationPreset

Generate report
report = Report(metrics=[DataDriftPreset(), ClassificationPreset()])
report.run(
    referencedata=refdf,
    currentdata=currdf,
    columnmapping=columnmapping
)

Tambahkan ke project
ws.addreport(project.id, report)

3. Start Monitoring UI

# Start Evidently UI
evidently ui --workspace myworkspace

4. Monitoring Over Time

import datetime

Tambahkan snapshots seiring waktu
for day in range(30):
    date = datetime.datetime(2024, 1, 1) + datetime.timedelta(days=day)
    dailydata = getdatafordate(date)

    report = Report(metrics=[DataDriftPreset()])
    report.run(referencedata=refdf, currentdata=dailydata)

    ws.addreport(project.id, report, timestamp=date)

Contoh Integrasi

1. Integrasi Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from datetime import datetime
import pandas as pd
from evidently.testsuite import TestSuite
from evidently.testpreset import DataDriftTestPreset


def validatedata(context):

    referencedata = pd.readparquet("s3://bucket/referencedata.parquet")

    currentdata = pd.readparquet(f"s3://bucket/data/{context['ds']}.parquet")


    suite = TestSuite(tests=[DataDriftTestPreset()])
    suite.run(referencedata=referencedata, currentdata=currentdata)


    # Save report
    suite.savehtml(f"/reports/driftreport{context['ds']}.html")

    if suite.asdict()["summary"]["allpassed"]:
        return "proceedwithtraining"
    else:
        return "alertteam"


def alertteam(context):
    # Kirim alert
    print("Data drift detected! Alerting team...")

dag = DAG(
    "modelmonitoring",

    startdate=datetime(2024, 1, 1),
    scheduleinterval="@daily",

)

validate = BranchPythonOperator(
    taskid="validatedata",

    pythoncallable=validatedata,

    dag=dag,
)

2. Integrasi FastAPI

from fastapi import FastAPI, HTTPException
from evidently.report import Report
from evidently.metricpreset import DataDriftPreset
import pandas as pd

app = FastAPI()
referencedata = pd.readcsv("referencedata.csv")


@app.post("/check-drift")
async def checkdrift(data: dict):
    currentdata = pd.DataFrame(data["records"])


    report = Report(metrics=[DataDriftPreset()])
    report.run(referencedata=referencedata, currentdata=currentdata)


    results = report.asdict()
    driftdetected = results["metrics"][0]["result"]["datasetdrift"]

    return {
        "driftdetected": driftdetected,
        "driftshare": results["metrics"][0]["result"]["shareofdriftedcolumns"],
    }

3. Integrasi MLflow

import mlflow
from evidently.report import Report
from evidently.metricpreset import RegressionPreset


def logmodelmetrics(model, Xtest, ytest):

    predictions = model.predict(Xtest)

    data = Xtest.copy()

    data["target"] = ytest
    data["prediction"] = predictions

    report = Report(metrics=[RegressionPreset()])
    report.run(currentdata=data)


    results = report.asdict()
    metrics = results["metrics"][0]["result"]["current"]

    with mlflow.startrun():

        mlflow.logmetric("mae", metrics["meanabserror"])
        mlflow.logmetric("rmse", metrics["rmse"])

        mlflow.logmetric("r2", metrics["r2score"])


        # Log report sebagai artifact
        report.savehtml("evidentlyreport.html")

        mlflow.logartifact("evidentlyreport.html")

4. Integrasi Grafana

from evidently.report import Report
from evidently.metricpreset import DataDriftPreset
from prometheusclient import Gauge, starthttpserver


Define Prometheus metrics
driftscore = Gauge("mldatadriftscore", "Data drift score")

driftedfeatures = Gauge("mldriftedfeaturescount", "Number of drifted features")


def updatemetrics(refdata, currdata):
    report = Report(metrics=[DataDriftPreset()])
    report.run(referencedata=refdata, currentdata=currdata)

    results = report.asdict()

    driftshare = results["metrics"][0]["result"]["shareofdriftedcolumns"]

    numdrifted = results["metrics"][0]["result"]["numberofdriftedcolumns"]


    driftscore.set(driftshare)

    driftedfeatures.set(numdrifted)


Start Prometheus server
starthttpserver(8000)

Production Monitoring Pipeline

from evidently.report import Report
from evidently.testsuite import TestSuite
from evidently.metricpreset import DataDriftPreset, ClassificationPreset

from evidently.testpreset import DataDriftTestPreset
import pandas as pd
import datetime
import json

class ModelMonitor:
    def init(self, referencedata: pd.DataFrame, modelname: str):
        self.referencedata = referencedata
        self.modelname = modelname
        self.alerts = []

    def checkdatadrift(self, currentdata: pd.DataFrame) -> dict:

        """Check untuk data drift"""
        report = Report(metrics=[DataDriftPreset()])
        report.run(
            referencedata=self.referencedata,

            currentdata=currentdata

        )

        results = report.asdict()
        driftdetected = results["metrics"][0]["result"]["datasetdrift"]

        if driftdetected:

            self.alerts.append({
                "type": "datadrift",
                "timestamp": datetime.datetime.now().isoformat(),
                "details": results["metrics"][0]["result"]
            })

        return results

    def checkmodelperformance(self, data: pd.DataFrame,
                                 columnmapping: dict) -> dict:

        """Check model performance"""
        report = Report(metrics=[ClassificationPreset()])
        report.run(currentdata=data, columnmapping=columnmapping)

        results = report.asdict()

        return results

    def runtests(self, currentdata: pd.DataFrame,

                  columnmapping: dict = None) -> bool:
        """Jalankan semua tests"""
        suite = TestSuite(tests=[DataDriftTestPreset()])
        suite.run(
            referencedata=self.referencedata,
            currentdata=currentdata,
            columnmapping=columnmapping
        )

        return suite.asdict()["summary"]["allpassed"]

    def generatereport(self, currentdata: pd.DataFrame,
                        outputpath: str):

        """Generate comprehensive report"""
        report = Report(metrics=[
            DataDriftPreset(),
            ClassificationPreset()
        ])
        report.run(
            referencedata=self.referencedata,

            currentdata=currentdata

        )
        report.savehtml(outputpath)


Usage
monitor = ModelMonitor(referencedata, "frauddetectionmodel")

Daily monitoring
for batch in dailybatches:

    results = monitor.checkdatadrift(batch)

    if not monitor.runtests(batch):
        sendalert("Tests failed!")


    monitor.generatereport(batch, f"reports/{date}.html")

Best Practices

1. Setup Reference Data

# Gunakan stable training data sebagai reference
referencedata = trainingdata.sample(n=10000, randomstate=42)

referencedata.toparquet("referencedata.parquet")

2. Define Custom Thresholds

from evidently.tests import TestColumnDrift

Adjust thresholds per column
suite = TestSuite(tests=[
    TestColumnDrift(
        columnname="criticalfeature",
        stattestthreshold=0.01,  # Stricter threshold

    ),
    TestColumnDrift(
        columnname="lessimportantfeature",
        stattestthreshold=0.1,  # More lenient

    ),
])

3. Schedule Regular Checks

# Jalankan daily/hourly checks
from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

@scheduler.scheduledjob('cron', hour=0)  # Daily di midnight
def dailymonitoring():

    currentdata = getlast24hdata()

    monitor.checkdatadrift(currentdata)
    monitor.generatereport(currentdata, f"reports/{date}.html")

scheduler.start()

Kesimpulan

Evidently adalah essential untuk ML monitoring dengan:

Data drift detection: Catch perubahan distribusi

Model performance tracking: Monitor metrics seiring waktu

Automated testing: CI/CD quality gates

Visual reports: Dashboard HTML interaktif

Easy integration: Works dengan ML stack apapun

Key takeaways:

• Setup reference data dari training
• Monitor baik data maupun model metrics
• Gunakan test suites untuk automated checks
• Set appropriate thresholds per feature
• Integrate dengan alerting systems