Split View: ML 모델 모니터링과 드리프트 탐지: Evidently AI + MLflow 프로덕션 운영 가이드

# Evidently AI installation (including MLflow integration)
# pip install evidently mlflow scikit-learn pandas

import pandas as pd
import numpy as np
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import (
    DatasetDriftMetric,
    DataDriftTable,
    ColumnDriftMetric,
)

# Prepare reference / current data
data = load_iris(as_frame=True)
df = data.frame
df.columns = ["sepal_length", "sepal_width", "petal_length", "petal_width", "target"]

reference_data = df.sample(frac=0.5, random_state=42)
current_data = df.drop(reference_data.index)

# Create simulated data with data drift
current_drifted = current_data.copy()
current_drifted["sepal_length"] = current_drifted["sepal_length"] + np.random.normal(2.0, 0.5, len(current_drifted))
current_drifted["petal_width"] = current_drifted["petal_width"] * 1.8

# Generate drift report
drift_report = Report(metrics=[
    DatasetDriftMetric(),
    DataDriftTable(),
])

drift_report.run(
    reference_data=reference_data,
    current_data=current_drifted,
)

# Extract results as dictionary (for programmatic use)
result = drift_report.as_dict()
dataset_drift = result["metrics"][0]["result"]["dataset_drift"]
drift_share = result["metrics"][0]["result"]["share_of_drifted_columns"]

print(f"Dataset drift detected: {dataset_drift}")
print(f"Drifted column ratio: {drift_share:.2%}")

# Save as HTML report
drift_report.save_html("drift_report.html")

from evidently.test_suite import TestSuite
from evidently.test_preset import DataDriftTestPreset, DataQualityTestPreset
from evidently.tests import (
    TestColumnDrift,
    TestShareOfDriftedColumns,
    TestNumberOfMissingValues,
    TestShareOfOutRangeValues,
    TestMeanInNSigmas,
)

# Configure data drift + quality test suite
monitoring_suite = TestSuite(tests=[
    # Drift test: fail if 30% or more of columns drift
    TestShareOfDriftedColumns(lt=0.3),

    # Individual key feature drift validation
    TestColumnDrift(column_name="sepal_length"),
    TestColumnDrift(column_name="petal_width"),

    # Data quality tests
    TestNumberOfMissingValues(eq=0),

    # Value range validation: sepal_length within +/- 3 sigma of reference data
    TestMeanInNSigmas(column_name="sepal_length", n=3),
])

monitoring_suite.run(
    reference_data=reference_data,
    current_data=current_drifted,
)

# Programmatically check test results
suite_result = monitoring_suite.as_dict()
all_passed = all(
    test["status"] == "SUCCESS"
    for test in suite_result["tests"]
)

print(f"All tests passed: {all_passed}")
for test in suite_result["tests"]:
    status_icon = "PASS" if test["status"] == "SUCCESS" else "FAIL"
    print(f"  [{status_icon}] {test['name']}: {test['status']}")

# Use as exit code in CI/CD pipelines
if not all_passed:
    print("ALERT: Data drift or quality anomaly detected. Retraining pipeline trigger required.")
    # sys.exit(1)  # Fail the build in CI

import mlflow
from evidently.report import Report
from evidently.metrics import (
    DatasetDriftMetric,
    DataDriftTable,
    ColumnDriftMetric,
)
import json
from datetime import datetime

# MLflow tracking server configuration
mlflow.set_tracking_uri("http://mlflow.internal:5000")
mlflow.set_experiment("model-monitoring/fraud-detection-v2")

def log_drift_to_mlflow(
    reference_data,
    current_data,
    model_name: str,
    model_version: str,
    batch_id: str,
):
    """Log drift analysis results to MLflow"""

    # Generate Evidently drift report
    drift_report = Report(metrics=[
        DatasetDriftMetric(),
        DataDriftTable(),
    ])
    drift_report.run(
        reference_data=reference_data,
        current_data=current_data,
    )

    result = drift_report.as_dict()
    drift_result = result["metrics"][0]["result"]

    # Record as MLflow Run
    with mlflow.start_run(run_name=f"drift-check-{batch_id}") as run:
        # Basic drift metrics
        mlflow.log_metric("dataset_drift_detected", int(drift_result["dataset_drift"]))
        mlflow.log_metric("drifted_columns_share", drift_result["share_of_drifted_columns"])
        mlflow.log_metric("number_of_drifted_columns", drift_result["number_of_drifted_columns"])
        mlflow.log_metric("total_columns", drift_result["number_of_columns"])

        # Log individual column drift scores
        column_drift = result["metrics"][1]["result"]["drift_by_columns"]
        for col_name, col_info in column_drift.items():
            safe_col_name = col_name.replace(" ", "_").replace("/", "_")
            mlflow.log_metric(
                f"drift_score_{safe_col_name}",
                col_info.get("drift_score", 0.0),
            )
            mlflow.log_metric(
                f"drift_detected_{safe_col_name}",
                int(col_info.get("column_drift", False)),
            )

        # Record metadata as tags
        mlflow.set_tags({
            "monitoring.type": "drift_detection",
            "monitoring.model_name": model_name,
            "monitoring.model_version": model_version,
            "monitoring.batch_id": batch_id,
            "monitoring.timestamp": datetime.utcnow().isoformat(),
            "monitoring.reference_size": str(len(reference_data)),
            "monitoring.current_size": str(len(current_data)),
        })

        # Save HTML report as artifact
        report_path = f"/tmp/drift_report_{batch_id}.html"
        drift_report.save_html(report_path)
        mlflow.log_artifact(report_path, artifact_path="drift_reports")

        # Save JSON results as artifact
        json_path = f"/tmp/drift_result_{batch_id}.json"
        with open(json_path, "w") as f:
            json.dump(result, f, indent=2, default=str)
        mlflow.log_artifact(json_path, artifact_path="drift_reports")

        print(f"Drift results logged to MLflow. Run ID: {run.info.run_id}")
        return drift_result["dataset_drift"], drift_result["share_of_drifted_columns"]


# Usage example
is_drifted, drift_share = log_drift_to_mlflow(
    reference_data=reference_data,
    current_data=current_drifted,
    model_name="fraud-detector",
    model_version="3",
    batch_id="2026-03-06-batch-001",
)

from mlflow import MlflowClient

client = MlflowClient(tracking_uri="http://mlflow.internal:5000")

MODEL_NAME = "fraud-detector"

def handle_drift_detection(
    is_drifted: bool,
    drift_share: float,
    model_name: str = MODEL_NAME,
    drift_threshold_warn: float = 0.2,
    drift_threshold_critical: float = 0.5,
):
    """Perform model registry actions based on drift detection results"""

    # Check current production model version
    try:
        prod_version = client.get_model_version_by_alias(model_name, "production")
        current_version = prod_version.version
        print(f"Current production model version: {current_version}")
    except Exception as e:
        print(f"Failed to retrieve production model alias: {e}")
        return

    if not is_drifted:
        print("No drift detected. Maintaining current model.")
        client.set_model_version_tag(
            model_name, current_version,
            key="last_drift_check",
            value="passed",
        )
        return

    if drift_share >= drift_threshold_critical:
        # Critical drift: immediately switch to fallback model + trigger retraining
        print(f"CRITICAL: Drift ratio {drift_share:.1%} - Switching to fallback model and triggering retraining")
        client.set_model_version_tag(
            model_name, current_version,
            key="drift_status", value="critical",
        )
        # Switch to fallback model if available
        try:
            fallback = client.get_model_version_by_alias(model_name, "fallback")
            client.set_registered_model_alias(model_name, "production", fallback.version)
            print(f"Switched to fallback model version {fallback.version}")
        except Exception:
            print("WARNING: No fallback model available. Maintaining current model while emergency retraining is needed.")

        # Trigger retraining (external system call)
        trigger_retraining(model_name, reason="critical_drift")

    elif drift_share >= drift_threshold_warn:
        # Warning level drift: record tag + notification
        print(f"WARNING: Drift ratio {drift_share:.1%} - Enhanced monitoring and scheduling retraining")
        client.set_model_version_tag(
            model_name, current_version,
            key="drift_status", value="warning",
        )
        # Add to scheduled retraining queue
        schedule_retraining(model_name, priority="normal")


def trigger_retraining(model_name: str, reason: str):
    """Trigger emergency retraining (call Airflow DAG, Kubeflow Pipeline, etc.)"""
    print(f"Retraining triggered: model={model_name}, reason={reason}")
    # requests.post("http://airflow.internal/api/v1/dags/retrain/dagRuns", ...)


def schedule_retraining(model_name: str, priority: str):
    """Register in scheduled retraining queue"""
    print(f"Retraining scheduled: model={model_name}, priority={priority}")


# Execute
handle_drift_detection(
    is_drifted=True,
    drift_share=0.55,
    model_name=MODEL_NAME,
)

# Airflow DAG example: drift check + conditional retraining
# dag_drift_monitor.py

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.dates import days_ago
from datetime import timedelta
import pandas as pd

default_args = {
    "owner": "ml-platform",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "execution_timeout": timedelta(minutes=30),
}

dag = DAG(
    dag_id="ml_drift_monitor_fraud_detection",
    default_args=default_args,
    description="Daily drift monitoring and conditional retraining",
    schedule_interval="0 6 * * *",  # Daily at 6 AM
    start_date=days_ago(1),
    catchup=False,
    tags=["ml-monitoring", "drift-detection"],
)


def fetch_data(**context):
    """Load reference data and last 24 hours of serving data"""
    from sqlalchemy import create_engine
    engine = create_engine("postgresql://reader:password@db.internal/features")

    reference = pd.read_sql(
        "SELECT * FROM fraud_features_reference", engine
    )
    current = pd.read_sql(
        """SELECT * FROM fraud_features_serving
           WHERE created_at >= NOW() - INTERVAL '24 hours'""",
        engine,
    )

    # Pass paths via XCom (store large data in S3)
    ref_path = "/tmp/reference_data.parquet"
    cur_path = "/tmp/current_data.parquet"
    reference.to_parquet(ref_path)
    current.to_parquet(cur_path)

    context["ti"].xcom_push(key="reference_path", value=ref_path)
    context["ti"].xcom_push(key="current_path", value=cur_path)
    context["ti"].xcom_push(key="current_size", value=len(current))


def run_drift_check(**context):
    """Run Evidently drift analysis and log to MLflow"""
    from evidently.report import Report
    from evidently.metrics import DatasetDriftMetric, DataDriftTable
    import mlflow

    ti = context["ti"]
    ref_path = ti.xcom_pull(key="reference_path")
    cur_path = ti.xcom_pull(key="current_path")

    reference = pd.read_parquet(ref_path)
    current = pd.read_parquet(cur_path)

    # Validate minimum sample count
    if len(current) < 100:
        print(f"Insufficient current data samples: {len(current)}. Skipping drift check.")
        ti.xcom_push(key="drift_action", value="skip")
        return "skip_retraining"

    report = Report(metrics=[DatasetDriftMetric(), DataDriftTable()])
    report.run(reference_data=reference, current_data=current)
    result = report.as_dict()

    drift_detected = result["metrics"][0]["result"]["dataset_drift"]
    drift_share = result["metrics"][0]["result"]["share_of_drifted_columns"]

    # Log to MLflow
    mlflow.set_tracking_uri("http://mlflow.internal:5000")
    mlflow.set_experiment("monitoring/fraud-detection")
    with mlflow.start_run(run_name=f"drift-{context['ds']}"):
        mlflow.log_metric("drift_detected", int(drift_detected))
        mlflow.log_metric("drift_share", drift_share)

    ti.xcom_push(key="drift_detected", value=drift_detected)
    ti.xcom_push(key="drift_share", value=drift_share)


def decide_action(**context):
    """Decide whether to retrain based on drift level"""
    ti = context["ti"]
    drift_detected = ti.xcom_pull(key="drift_detected")
    drift_share = ti.xcom_pull(key="drift_share")

    if drift_share is None or drift_share < 0.2:
        return "skip_retraining"
    elif drift_share >= 0.5:
        return "trigger_emergency_retrain"
    else:
        return "trigger_scheduled_retrain"


fetch_task = PythonOperator(
    task_id="fetch_data", python_callable=fetch_data, dag=dag,
)
drift_task = PythonOperator(
    task_id="run_drift_check", python_callable=run_drift_check, dag=dag,
)
branch_task = BranchPythonOperator(
    task_id="decide_action", python_callable=decide_action, dag=dag,
)
skip_task = EmptyOperator(task_id="skip_retraining", dag=dag)
scheduled_retrain = EmptyOperator(task_id="trigger_scheduled_retrain", dag=dag)
emergency_retrain = EmptyOperator(task_id="trigger_emergency_retrain", dag=dag)

fetch_task >> drift_task >> branch_task >> [skip_task, scheduled_retrain, emergency_retrain]

# prometheus_drift_exporter.py
from prometheus_client import start_http_server, Gauge, Counter, Histogram
from evidently.report import Report
from evidently.metrics import DatasetDriftMetric, DataDriftTable
import pandas as pd
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Prometheus metric definitions
DRIFT_DETECTED = Gauge(
    "ml_model_drift_detected",
    "Dataset drift detection status (0/1)",
    ["model_name", "model_version"],
)
DRIFT_SHARE = Gauge(
    "ml_model_drift_column_share",
    "Share of drifted columns",
    ["model_name", "model_version"],
)
COLUMN_DRIFT_SCORE = Gauge(
    "ml_model_column_drift_score",
    "Individual column drift score",
    ["model_name", "model_version", "column_name"],
)
DRIFT_CHECK_TOTAL = Counter(
    "ml_model_drift_checks_total",
    "Total drift check executions",
    ["model_name"],
)
DRIFT_CHECK_DURATION = Histogram(
    "ml_model_drift_check_duration_seconds",
    "Drift check execution duration",
    ["model_name"],
    buckets=[0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0],
)

MODEL_NAME = "fraud-detector"
MODEL_VERSION = "3"


def run_periodic_drift_check(
    reference_path: str,
    current_query_fn,
    interval_seconds: int = 300,
):
    """Periodic drift check and Prometheus metric update"""
    reference = pd.read_parquet(reference_path)

    while True:
        try:
            start_time = time.time()

            # Load recent data
            current = current_query_fn()
            if current is None or len(current) < 50:
                logger.warning(f"Insufficient current data: {len(current) if current is not None else 0} records")
                time.sleep(interval_seconds)
                continue

            # Filter to feature columns only (exclude target and metadata columns)
            feature_cols = [c for c in reference.columns if c not in ["target", "id", "timestamp"]]
            ref_features = reference[feature_cols]
            cur_features = current[feature_cols]

            # Drift analysis
            report = Report(metrics=[DatasetDriftMetric(), DataDriftTable()])
            report.run(reference_data=ref_features, current_data=cur_features)
            result = report.as_dict()

            drift_result = result["metrics"][0]["result"]
            column_results = result["metrics"][1]["result"]["drift_by_columns"]

            # Update Prometheus metrics
            DRIFT_DETECTED.labels(MODEL_NAME, MODEL_VERSION).set(
                int(drift_result["dataset_drift"])
            )
            DRIFT_SHARE.labels(MODEL_NAME, MODEL_VERSION).set(
                drift_result["share_of_drifted_columns"]
            )

            for col_name, col_info in column_results.items():
                COLUMN_DRIFT_SCORE.labels(MODEL_NAME, MODEL_VERSION, col_name).set(
                    col_info.get("drift_score", 0.0)
                )

            DRIFT_CHECK_TOTAL.labels(MODEL_NAME).inc()

            duration = time.time() - start_time
            DRIFT_CHECK_DURATION.labels(MODEL_NAME).observe(duration)

            logger.info(
                f"Drift check complete: drift={drift_result['dataset_drift']}, "
                f"share={drift_result['share_of_drifted_columns']:.2%}, "
                f"duration={duration:.1f}s"
            )

        except Exception as e:
            logger.error(f"Drift check failed: {e}", exc_info=True)

        time.sleep(interval_seconds)


if __name__ == "__main__":
    # Start Prometheus metrics HTTP server (port 8000)
    start_http_server(8000)
    logger.info("Prometheus metrics exporter started (port 8000)")

    # Start periodic drift check (5-minute intervals)
    run_periodic_drift_check(
        reference_path="/data/reference/fraud_features_v3.parquet",
        current_query_fn=lambda: pd.read_parquet("/data/serving/latest_batch.parquet"),
        interval_seconds=300,
    )

# prometheus-alerts.yaml
groups:
  - name: ml_model_drift_alerts
    rules:
      - alert: MLModelDriftDetected
        expr: ml_model_drift_detected == 1
        for: 5m
        labels:
          severity: warning
          team: ml-platform
        annotations:
          summary: 'ML model data drift detected'
          description: 'Data drift detected in model {{ $labels.model_name }} v{{ $labels.model_version }}. Drifted column ratio: {{ $value }}'

      - alert: MLModelCriticalDrift
        expr: ml_model_drift_column_share > 0.5
        for: 0m
        labels:
          severity: critical
          team: ml-platform
        annotations:
          summary: 'ML model critical drift - Immediate action required'
          description: 'Model {{ $labels.model_name }} drift column ratio is {{ $value | humanizePercentage }}. Immediate retraining or fallback switch is required.'

      - alert: MLDriftCheckStalled
        expr: rate(ml_model_drift_checks_total[1h]) == 0
        for: 30m
        labels:
          severity: warning
          team: ml-platform
        annotations:
          summary: 'ML drift check stalled'
          description: 'Drift check for model {{ $labels.model_name }} has not run for over 30 minutes. Monitoring pipeline inspection is needed.'