Complete Evidently AI Tutorial: ML Model Monitoring and Data Quality

Evidently is an open-source Python library for evaluating, testing, and monitoring machine learning models in production. It helps detect data drift, model degradation, and data quality issues before they impact your business.

Why Evidently?

Evidently Advantages:

• Data drift detection: Monitor input data changes
• Model performance tracking: Track metrics over time
• Visual reports: Interactive HTML dashboards
• Test suites: Automated quality checks
• Easy integration: Works with any ML framework

Use Cases:

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

Installation

# Basic installation
pip install evidently

With 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) and 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 as HTML
report.savehtml("driftreport.html")


Get results as 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 with predictions and 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 for 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 for:
- 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  # For 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("All tests passed!")
else:
    print("Some 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,  # No negative quantities
    ),
])

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",  # or "regression"
)

Use in 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 for recommendation model"
project.save()

2. Add Reports to 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
)

Add to project
ws.addreport(project.id, report)

3. Start Monitoring UI

# Start Evidently UI
evidently ui --workspace myworkspace

4. Monitoring Over Time

import datetime

Add snapshots over time
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)

Integration Examples

1. Airflow Integration

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):
    # Send 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. FastAPI Integration

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. MLflow Integration

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 as artifact
        report.savehtml("evidentlyreport.html")

        mlflow.logartifact("evidentlyreport.html")

4. Grafana Integration

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 for 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:
        """Run all 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. Set Up Reference Data

# Use stable training data as 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

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

scheduler = BackgroundScheduler()

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

    currentdata = getlast24hdata()

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

scheduler.start()

Conclusion

Evidently is essential for ML monitoring with:

Data drift detection: Catch distribution changes

Model performance tracking: Monitor metrics over time

Automated testing: CI/CD quality gates

Visual reports: Interactive HTML dashboards

Easy integration: Works with any ML stack

Key takeaways:

• Set up reference data from training
• Monitor both data and model metrics
• Use test suites for automated checks
• Set appropriate thresholds per feature
• Integrate with alerting systems