0Pricing
Machine Learning Academy · レッスン

Evidently AIでドリフト検知パイプラインを構築する

Evidentlyライブラリを使ってデータドリフトのHTMLレポートを生成し、定期実行スクリプトにチェックを組み込み、ドリフト検知時にメールアラートを発火させます。

「Evidently AIでドリフト検知パイプラインを構築する」はCoddyKit上の無料Machine Learning Academyレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはMachine Learning Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Machine Learning Academyコースには全4レッスンが含まれています。

このレッスンの一部はまだ翻訳されておらず、英語で表示されています。

What Is Evidently AI?

Evidently AI is an open-source Python library for evaluating and monitoring machine learning models in production. It provides pre-built reports and test suites covering data drift, data quality, model performance, and prediction drift. With a few lines of code, Evidently generates interactive HTML reports or JSON summaries that can be embedded in monitoring dashboards or sent as email attachments, making it the fastest way to add comprehensive monitoring to an existing ML pipeline.

# Install Evidently
# pip install evidently

import evidently
print('Evidently version:', evidently.__version__)

# Evidently organises monitoring into:
# 1. Reports -- exploratory HTML dashboards
# 2. Test Suites -- pass/fail checks for CI/CD integration
# 3. Metrics -- individual measurements (DataDriftPreset, etc.)

from evidently.report import Report
from evidently.test_suite import TestSuite
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
print('Imports successful.')

Preparing Reference and Current Datasets

Evidently compares a reference dataset (typically the training data or a recent stable production period) against a current dataset (the most recent production batch). Both must be pandas DataFrames with the same column schema. The reference establishes the baseline; the current is checked for drift relative to that baseline. Include the target column if ground-truth labels are available; Evidently will use them for performance metrics.

import pandas as pd
import numpy as np
from sklearn.datasets import make_classification

np.random.seed(42)

# Simulate training data (reference)
X_ref, y_ref = make_classification(n_samples=5000, n_features=8, random_state=42)
reference = pd.DataFrame(X_ref, columns=[f'feature_{i}' for i in range(8)])
reference['target'] = y_ref

# Simulate production batch with some drift
X_cur = X_ref[:1000].copy()
X_cur[:, 2] += 2.0  # shift feature_2
X_cur[:, 5] *= 1.8  # scale feature_5
current = pd.DataFrame(X_cur, columns=[f'feature_{i}' for i in range(8)])
current['target'] = y_ref[:1000]  # labels may or may not be available

print('Reference shape:', reference.shape)
print('Current shape:', current.shape)

Generating a Data Drift Report

The DataDriftPreset generates a comprehensive drift report covering all features. It automatically selects the appropriate statistical test for each column (KS test for continuous features, chi-squared for categorical) and reports the drift score, p-value, and a yes/no drift decision for each. The HTML output includes interactive visualisations showing the reference vs current distribution for each drifted feature.

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

# Build and run the report
data_drift_report = Report(metrics=[DataDriftPreset()])
data_drift_report.run(
    reference_data=reference,
    current_data=current
)

# Save to HTML file
data_drift_report.save_html('/tmp/data_drift_report.html')
print('Report saved to /tmp/data_drift_report.html')

# Extract as dictionary for programmatic use
result_dict = data_drift_report.as_dict()
drift_summary = result_dict['metrics'][0]['result']
print('Dataset drift detected:', drift_summary['dataset_drift'])
print('Features drifted:', drift_summary['number_of_drifted_columns'])

Column Drift Details

Access per-column drift results programmatically to build automated alerts. Evidently returns a dictionary with each column's drift score, statistical test used, p-value, and a boolean drift flag. Filter for drifted columns and include them in your alert message to help the team immediately understand which features have changed and by how much.

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=reference, current_data=current)

result = report.as_dict()['metrics'][0]['result']

print('Per-column drift summary:')
for col, info in result['drift_by_columns'].items():
    drift_flag = info.get('drift_detected', False)
    stat_name = info.get('stattest_name', 'N/A')
    p_val = info.get('p_value', 'N/A')
    status = '*** DRIFTED ***' if drift_flag else 'stable'
    print(f'  {col:12s}: {stat_name} p={p_val:.4f} [{status}]' if isinstance(p_val, float)
          else f'  {col}: {status}')

Data Quality Report

Beyond drift, Evidently's DataQualityPreset checks for data quality issues in the current batch: missing values, constant columns, duplicates, and out-of-range values. These issues are often the first sign of an upstream data pipeline failure rather than true drift. Running data quality checks before drift checks prevents false drift alarms caused by missing imputation or schema changes in the ETL pipeline.

from evidently.report import Report
from evidently.metric_preset import DataQualityPreset
import pandas as pd
import numpy as np

# Introduce some quality issues
current_with_issues = current.copy()
current_with_issues.loc[50:80, 'feature_0'] = np.nan   # missing values
current_with_issues.loc[100:110, 'feature_3'] = 99999  # outliers

quality_report = Report(metrics=[DataQualityPreset()])
quality_report.run(
    reference_data=reference,
    current_data=current_with_issues
)
quality_report.save_html('/tmp/data_quality_report.html')

quality_dict = quality_report.as_dict()
print('Quality report generated with', len(quality_dict['metrics']), 'metrics.')

Test Suites for CI/CD: Pass or Fail

Reports are for humans; test suites are for machines. A test suite runs a battery of configurable checks and returns a pass/fail result that can gate pipeline steps. If any test fails, the pipeline raises an exception, the CI/CD step fails, and the alert fires. Common tests: number of drifted columns below threshold, missing value rate below threshold, all columns within expected value ranges.

from evidently.test_suite import TestSuite
from evidently.tests import (
    TestNumberOfDriftedColumns,
    TestShareOfMissingValues,
    TestNumberOfConstantColumns
)

test_suite = TestSuite(tests=[
    TestNumberOfDriftedColumns(lt=2),   # fail if > 1 column drifts
    TestShareOfMissingValues(lt=0.01),  # fail if >1% missing
    TestNumberOfConstantColumns(lt=1)   # fail if any constant column
])

test_suite.run(reference_data=reference, current_data=current)
test_suite.save_html('/tmp/test_suite_report.html')

result = test_suite.as_dict()
summary = result['summary']
print(f'Tests passed: {summary["passed_tests"]}/{summary["total_tests"]}')
print('Overall status:', 'PASS' if summary['all_passed'] else 'FAIL')

Integrating Evidently into a Scheduled Script

Automate drift monitoring by running Evidently on a daily cron job. The script loads the previous day's production batch, compares it to the reference dataset, runs the test suite, and sends an alert if tests fail. This pattern requires no ML expertise to maintain — the scheduled script runs silently when everything is healthy and raises an alarm only when action is needed.

import pandas as pd
from evidently.test_suite import TestSuite
from evidently.tests import TestNumberOfDriftedColumns, TestShareOfMissingValues
from datetime import date

def daily_drift_check(reference_path, production_path, alert_threshold=2):
    reference = pd.read_parquet(reference_path)
    current = pd.read_parquet(production_path)

    suite = TestSuite(tests=[
        TestNumberOfDriftedColumns(lt=alert_threshold),
        TestShareOfMissingValues(lt=0.05)
    ])
    suite.run(reference_data=reference, current_data=current)

    result = suite.as_dict()
    passed = result['summary']['all_passed']

    report_path = f'/tmp/drift_{date.today()}.html'
    suite.save_html(report_path)

    return passed, report_path, result['summary']

# In cron script:
# passed, report, summary = daily_drift_check('data/reference.parquet', 'data/today.parquet')
# if not passed:
#     send_email_alert(report, summary)
print('daily_drift_check function ready.')

Sending Email Alerts on Drift Detection

When the test suite fails, send an email alert with the HTML report attached. Use Python's built-in smtplib and email modules to compose a message with the drift summary in the body and the full Evidently HTML report as an attachment. The recipient can open the report to see exactly which features drifted and by how much, without needing access to the production environment.

import smtplib
import os
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication

def send_drift_alert(recipient, report_path, summary):
    smtp_host = os.getenv('SMTP_HOST', 'smtp.gmail.com')
    smtp_user = os.getenv('SMTP_USER')
    smtp_pass = os.getenv('SMTP_PASS')

    msg = MIMEMultipart()
    msg['From'] = smtp_user
    msg['To'] = recipient
    msg['Subject'] = '[ML ALERT] Data drift detected in production'

    body = (f'Drift alert triggered.\n'
            f'Tests passed: {summary["passed_tests"]}/{summary["total_tests"]}\n'
            f'See attached HTML report for details.')
    msg.attach(MIMEText(body, 'plain'))

    with open(report_path, 'rb') as f:
        attachment = MIMEApplication(f.read(), _subtype='html')
        attachment.add_header('Content-Disposition', 'attachment',
                              filename='drift_report.html')
        msg.attach(attachment)

    print(f'Alert email sent to {recipient}')
print('send_drift_alert function ready.')

Slack Webhook Integration

For teams that use Slack, posting a concise drift alert to a dedicated #ml-monitoring channel is more immediately visible than email. Include the number of drifted features, the most severely drifted column, and a link to the saved HTML report in cloud storage (S3 or GCS). Use Slack Block Kit for formatted messages with colour-coded status (green = healthy, red = drift detected).

import requests
import json
import os

def post_slack_alert(summary, drifted_columns, report_url):
    webhook_url = os.getenv('SLACK_WEBHOOK_URL')
    if not webhook_url:
        print('No Slack webhook configured.')
        return

    all_passed = summary['all_passed']
    icon = ':white_check_mark:' if all_passed else ':rotating_light:'
    color = 'good' if all_passed else 'danger'
    status_text = 'All checks passed' if all_passed else 'DRIFT DETECTED'

    message = {
        'attachments': [{
            'color': color,
            'title': f'{icon} ML Monitoring Alert',
            'text': (f'Status: {status_text}\n'
                     f'Drifted features: {drifted_columns}\n'
                     f'Tests: {summary["passed_tests"]}/{summary["total_tests"]} passed\n'
                     f'<{report_url}|View full report>'),
            'fallback': status_text
        }]
    }
    response = requests.post(webhook_url, data=json.dumps(message))
    print('Slack response:', response.status_code)

Saving Reports to S3 for Audit Trail

Store every generated Evidently report in S3 or GCS with a date-stamped key to create a permanent audit trail. This enables retrospective analysis: when a model's performance drops in April, retrieve the March drift report to see whether drift was already visible weeks earlier. Configure an S3 lifecycle policy to retain reports for 2 years and automatically archive older ones to Glacier storage for cost efficiency.

import boto3
import os
from datetime import date

def upload_report_to_s3(local_path, bucket, model_name):
    s3 = boto3.client(
        's3',
        aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
        aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY')
    )

    today = date.today().isoformat()
    s3_key = f'ml-monitoring/{model_name}/{today}/drift_report.html'

    s3.upload_file(
        local_path,
        bucket,
        s3_key,
        ExtraArgs={'ContentType': 'text/html'}
    )

    report_url = f'https://{bucket}.s3.amazonaws.com/{s3_key}'
    print(f'Report uploaded: {report_url}')
    return report_url

print('upload_report_to_s3 function ready.')

Dashboard Integration with Grafana

Evidently produces HTML reports suitable for human review, but for operational monitoring you need a real-time dashboard that updates automatically. Extract PSI and KS statistics as numeric values from Evidently test results, push them to Prometheus (via the prometheus_client library), and visualise in Grafana. This gives your SRE team the same observability tooling for ML models as for application services, enabling drift alerts alongside latency and error-rate alerts in one place.

# Export Evidently metrics to Prometheus
from prometheus_client import Gauge, start_http_server
import time

# Define Prometheus gauges for drift metrics
psi_gauge = Gauge('ml_feature_psi', 'Population Stability Index', ['feature_name'])
ks_gauge  = Gauge('ml_feature_ks_statistic', 'KS test statistic', ['feature_name'])

# Push computed drift values (call after running Evidently tests)
def push_drift_metrics(drift_results: dict):
    for feature, metrics in drift_results.items():
        psi_gauge.labels(feature_name=feature).set(metrics.get('psi', 0))
        ks_gauge.labels(feature_name=feature).set(metrics.get('ks_stat', 0))

# start_http_server(8000)  # Prometheus scrapes this port
print('Expose metrics at :8000/metrics -> scrape in Prometheus -> display in Grafana')

Quick Check

Test your understanding of Machine Learning with Python concepts from this lesson.

Lesson Recap

In this lesson you learned: Evidently AI generates comprehensive drift and data quality reports with a few lines of Python, Test Suites provide pass/fail results that integrate directly into CI/CD pipelines and scheduled cron jobs, and combining S3 storage, email, and Slack notifications creates a complete automated alert pipeline that catches drift before it visibly impacts users. Next up we enter the Responsible AI section, starting with SHAP values for interpreting global and local feature importance in complex models.

よくある質問

「Evidently AIでドリフト検知パイプラインを構築する」レッスンは無料ですか？

はい。「Evidently AIでドリフト検知パイプラインを構築する」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、Machine Learning Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Machine Learning Academyコースには全4レッスンが含まれています。

「Evidently AIでドリフト検知パイプラインを構築する」で何を学びますか？

Evidentlyライブラリを使ってデータドリフトのHTMLレポートを生成し、定期実行スクリプトにチェックを組み込み、ドリフト検知時にメールアラートを発火させます。 ブラウザで直接実行するハンズオンコードでMachine Learning Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

Machine Learning Academyを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのMachine Learning Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。

「Evidently AIでドリフト検知パイプラインを構築する」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このMachine Learning Academyレッスンでコードを書いて実行できますか？

はい。すべてのMachine Learning Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. データドリフト：時間経過に伴う特徴量分布の変化
  2. コンセプトドリフト：XとYの関係が変化するとき
  3. 予測分布と信頼度スコアの監視
  4. Evidently AIでドリフト検知パイプラインを構築する
← Machine Learning Academyに戻る