Моніторинг якості даних

Моніторинг якості даних забезпечує що дані для ML систем залишаються надійними та консистентними. Погані дані викликають погані predictions - garbage in, garbage out. На відміну від моніторингу моделей який ловить деградацію моделі, моніторинг якості даних ловить проблеми на джерелі перед їх поширенням через pipelines.

Де використовується

Якість даних критична скрізь де ML системи споживають дані:

• Feature pipelines: Валідація inputs до feature stores
• Training pipelines: Забезпечення якості training data
• Inference pipelines: Валідація real-time inputs
• Data warehouses: Моніторинг якості source data
• ETL процеси: Catching issues під час transformation
• Data contracts: Enforcement agreements між командами

Продукти: Great Expectations, dbt tests, Monte Carlo, Soda, Anomalo.

Виміри якості даних

Типові точки моніторингу

Що ви робите на практиці

1. Map data flows - Зрозумійте sources, transformations, consumers
2. Визначте expectations - Як кожен dataset повинен виглядати
3. Імплементуйте schema validation - Catch structural changes
4. Додайте statistical checks - Distributions, ranges, correlations
5. Налаштуйте freshness monitoring - Дані приходять вчасно
6. Configure alerting - Right people, right severity
7. Побудуйте data contracts - Agreements між producers/consumers
8. Створіть dashboards - Visibility в data health
9. Automate remediation - Де безпечно це робити
10. Review та iterate - Update expectations як data еволюціонує

Schema Validation

from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import pandas as pd

@dataclass
class ColumnSchema:
    name: str
    dtype: str
    nullable: bool = True
    min_value: Optional[float] = None
    max_value: Optional[float] = None
    allowed_values: Optional[List[Any]] = None

class SchemaValidator:
    def __init__(self, columns: List[ColumnSchema]):
        self.columns = {col.name: col for col in columns}

    def validate(self, df: pd.DataFrame) -> Dict[str, List[str]]:
        errors = {"missing_columns": [], "extra_columns": [], 
                  "type_errors": [], "constraint_errors": []}

        # Перевірка missing/extra columns
        expected = set(self.columns.keys())
        actual = set(df.columns)
        errors["missing_columns"] = list(expected - actual)
        errors["extra_columns"] = list(actual - expected)

        # Валідація кожної колонки
        for col_name, schema in self.columns.items():
            if col_name not in df.columns:
                continue

            col = df[col_name]

            # Null check
            if not schema.nullable and col.isna().any():
                errors["constraint_errors"].append(
                    f"{col_name}: contains nulls but nullable=False"
                )

            # Range check
            if schema.min_value is not None:
                if (col < schema.min_value).any():
                    errors["constraint_errors"].append(
                        f"{col_name}: values below minimum {schema.min_value}"
                    )

        return errors

Статистичний моніторинг

import numpy as np
from typing import Dict

class StatisticalMonitor:
    def __init__(self, reference_stats: Dict[str, Dict]):
        self.reference = reference_stats

    @staticmethod
    def compute_stats(df, columns: list) -> Dict[str, Dict]:
        """Compute reference statistics з baseline data."""
        stats_dict = {}
        for col in columns:
            if df[col].dtype in ['float64', 'int64']:
                stats_dict[col] = {
                    "mean": df[col].mean(),
                    "std": df[col].std(),
                    "min": df[col].min(),
                    "max": df[col].max(),
                    "null_rate": df[col].isna().mean()
                }
        return stats_dict

    def check_drift(self, df, threshold: float = 0.1) -> Dict[str, Dict]:
        """Порівняння поточних даних з reference statistics."""
        alerts = {}

        for col, ref_stats in self.reference.items():
            if col not in df.columns:
                alerts[col] = {"error": "column missing"}
                continue

            if "mean" in ref_stats:  # Numeric column
                current_mean = df[col].mean()
                ref_mean = ref_stats["mean"]
                ref_std = ref_stats["std"]

                # Z-score of mean shift
                if ref_std > 0:
                    mean_shift = abs(current_mean - ref_mean) / ref_std
                    if mean_shift > threshold * 3:
                        alerts[col] = {
                            "type": "mean_shift",
                            "reference": ref_mean,
                            "current": current_mean,
                            "z_score": mean_shift
                        }

        return alerts

    def psi(self, expected: np.ndarray, actual: np.ndarray, bins: int = 10) -> float:
        """Population Stability Index."""
        breakpoints = np.percentile(expected, np.linspace(0, 100, bins + 1))
        breakpoints[0] = -np.inf
        breakpoints[-1] = np.inf

        expected_counts = np.histogram(expected, breakpoints)[0] / len(expected)
        actual_counts = np.histogram(actual, breakpoints)[0] / len(actual)

        expected_counts = np.clip(expected_counts, 0.001, None)
        actual_counts = np.clip(actual_counts, 0.001, None)

        psi_value = np.sum((actual_counts - expected_counts) * 
                          np.log(actual_counts / expected_counts))
        return psi_value

Freshness Monitoring

from datetime import datetime, timedelta

class FreshnessMonitor:
    def __init__(self, expected_frequency: timedelta, 
                 max_delay: timedelta):
        self.expected_frequency = expected_frequency
        self.max_delay = max_delay
        self.last_arrival = None

    def record_arrival(self, timestamp: datetime = None):
        """Запис arrival даних."""
        self.last_arrival = timestamp or datetime.utcnow()

    def check_freshness(self) -> dict:
        """Перевірка чи дані fresh."""
        now = datetime.utcnow()

        if self.last_arrival is None:
            return {"status": "error", "message": "No data arrivals recorded"}

        age = now - self.last_arrival
        expected_next = self.last_arrival + self.expected_frequency

        if now > expected_next + self.max_delay:
            return {
                "status": "stale",
                "age": str(age),
                "delay": str(now - expected_next)
            }
        elif now > expected_next:
            return {"status": "delayed", "age": str(age)}
        else:
            return {"status": "fresh", "age": str(age)}

Data Contracts

from dataclasses import dataclass
from typing import List, Dict

@dataclass
class DataContract:
    """Agreement між data producer і consumer."""
    name: str
    producer: str
    consumers: List[str]
    schema: List[ColumnSchema]
    freshness_sla: timedelta
    quality_thresholds: Dict[str, float]

class ContractEnforcer:
    def __init__(self, contracts: List[DataContract]):
        self.contracts = {c.name: c for c in contracts}

    def validate_contract(self, contract_name: str, df, 
                          arrival_time: datetime) -> Dict:
        """Validate data against its contract."""
        contract = self.contracts[contract_name]
        results = {"contract": contract_name, "violations": []}

        # Schema validation
        validator = SchemaValidator(contract.schema)
        schema_errors = validator.validate(df)
        for error_type, errors in schema_errors.items():
            if errors:
                results["violations"].append({
                    "type": "schema",
                    "details": {error_type: errors}
                })

        results["valid"] = len(results["violations"]) == 0
        return results

Метрики оцінки

Типові помилки та чек-лист

• [ ] Моніторинг занадто пізно: Catching issues після того як модель використала bad data
• [ ] Alert fatigue: Занадто багато alerts, реальні issues ігноруються
• [ ] Static thresholds: Не адаптуються до сезонних patterns
• [ ] Missing freshness: Не моніторинг timeliness даних
• [ ] Schema drift ignored: Нові columns ламають pipelines тихо

Чек-лист для продакшену

Перед деплоєм: - [ ] Baseline statistics computed - [ ] Schema expectations defined - [ ] Freshness SLAs established - [ ] Alerting configured

Після деплою: - [ ] Dashboard visibility - [ ] Alert response procedures - [ ] Regular threshold review - [ ] Contract compliance tracking

Інструменти та стек

Приклад міні-кейсу

Сценарій: Data quality для e-commerce рекомендаційної системи.

Підхід: - Schema validation на ingestion - Statistical monitoring для price, inventory - 15-хвилинний freshness SLA - Data contracts з upstream командами

Результати: - 98% issue detection before model impact - 4 хвилини average detection time - 75% зменшення data incidents

Пов'язані сторінки

• Індекс: ML Field Guide
• Feature Stores
• Model Monitoring
• Розмітка даних

Додаткові матеріали

• Great Expectations Docs
• Data Contracts Specification
• Fundamentals of Data Engineering

Спробуйте наживо

Дослідіть пов'язану модель на практиці: відкрийте Building the Data Foundation for Machine Learning: A Data Quality Score Framework і змінюйте параметри під час роботи. Усе лишається у вашому браузері.