Как измерять и улучшать качество данных (data quality)?

Измерение и улучшение качества данных (Data Quality)

Data Quality — это степень соответствия данных требованиям бизнеса и техники. Плохое качество данных приводит к неверным аналитическим выводам, неправильным бизнес-решениям и потере доверия к аналитике. В крупных компаниях низкое качество данных стоит $12-15m в год (по Gartner).

Шесть измерений качества данных

1. Completeness (Полнота) — все ли нужные данные есть?

Процент непустых значений в важных полях.

-- KPI: Completeness
SELECT 
    column_name,
    COUNT(*) as total_rows,
    COUNT(CASE WHEN column_name IS NOT NULL THEN 1 END) as non_null_rows,
    ROUND(100.0 * COUNT(CASE WHEN column_name IS NOT NULL THEN 1 END) / COUNT(*), 2) as completeness_pct
FROM table_name
GROUP BY column_name;

-- Примеры:
-- user_email: 98% (хорошо)
-- phone: 45% (плохо)
-- payment_method: 100% (идеально, но может быть артефакт логики)

Пороги качества:

• 95-100%: ✅ Отличное
• 85-95%: ⚠️ Приемлемое
• < 85%: ❌ Плохое

Как улучшить: проверить, почему пусты значения. Может быть, поле не обязательное в форме? Или есть ошибка отслеживания?

2. Accuracy (Точность) — соответствуют ли данные реальности?

Процент значений, соответствующих правилам валидации.

-- KPI: Accuracy (примеры)
SELECT 
    COUNT(*) as total_transactions,
    COUNT(CASE WHEN amount > 0 THEN 1 END) as valid_amounts,
    COUNT(CASE WHEN currency IN ('USD', 'EUR', 'RUB') THEN 1 END) as valid_currency,
    COUNT(CASE WHEN transaction_date <= NOW() THEN 1 END) as non_future_dates,
    
    -- Accuracy score (среднее по всем чекам)
    ROUND(100.0 * (COUNT(CASE WHEN amount > 0 AND 
                                currency IN ('USD', 'EUR', 'RUB') AND 
                                transaction_date <= NOW() THEN 1 END)) / COUNT(*), 2) as accuracy_pct
FROM transactions;

Примеры ошибок точности:

• Отрицательные суммы заказов (amount < 0) ❌
• Будущие даты в прошлых событиях ❌
• Некорректные email'ы (@ отсутствует) ❌
• Статусы заказов со случайными символами ❌

Как улучшить: добавить валидацию на уровне приложения и БД.

3. Consistency (Консистентность) — соответствуют ли разные источники друг другу?

Одно и то же значение одинаково во всех системах.

# Пример: amount заказа в таблице orders должен совпадать с суммой items

import pandas as pd

orders = pd.read_sql("SELECT id, total_amount FROM orders")
items = pd.read_sql("SELECT order_id, SUM(price * qty) as calculated_amount FROM order_items GROUP BY order_id")

# Сравнение
merged = orders.merge(items, left_on='id', right_on='order_id')
mismatches = merged[merged['total_amount'] != merged['calculated_amount']]
consistency_pct = 100 * (1 - len(mismatches) / len(merged))

print(f"Consistency: {consistency_pct:.1f}%")
print(mismatches.head())  # Какие заказы не совпадают

Примеры несогласованности:

• user_country в таблице users ≠ страна доставки в заказе ❌
• revenue_total в BI системе ≠ SUM(transactions.amount) в БД ❌
• user.subscription_status = 'active', но user.last_payment дата > 90 дней ❌

Как улучшить: реплицировать данные с единым source of truth, добавить триггеры в БД.

4. Uniqueness (Уникальность) — нет ли дубликатов?

Каждый объект представлен один раз в указанном контексте.

-- KPI: Uniqueness
SELECT 
    COUNT(*) as total_rows,
    COUNT(DISTINCT user_id) as unique_users,
    ROUND(100.0 * COUNT(DISTINCT user_id) / COUNT(*), 2) as uniqueness_pct
FROM events;

-- Примеры проблем:
-- user_id = '123' встречается 10 раз в sessions (ошибка)
-- email дубликаты в users (нужна UNIQUE constraint)

-- Найти дубликаты
SELECT 
    user_id,
    COUNT(*) as occurrences
FROM users
GROUP BY user_id
HAVING COUNT(*) > 1;

Пороги: 99-100% уникальности ✅

Как улучшить:

-- Добавить ограничения
ALTER TABLE users 
ADD CONSTRAINT unique_email UNIQUE(email);

ALTER TABLE users 
ADD CONSTRAINT unique_phone UNIQUE(phone);

-- Дедупликация существующих
DELETE FROM users a
USING users b
WHERE a.id > b.id
AND a.email = b.email;

5. Timeliness (Своевременность) — актуальны ли данные?

Время, которое прошло с момента события до появления в BI.

-- KPI: Timeliness
SELECT 
    table_name,
    MAX(event_time) as last_event_time,
    NOW() - MAX(event_time) as lag,
    CASE 
        WHEN NOW() - MAX(event_time) < INTERVAL '1 hour' THEN 'Real-time ✅'
        WHEN NOW() - MAX(event_time) < INTERVAL '1 day' THEN 'Daily ⚠️'
        ELSE 'Stale ❌'
    END as timeliness_status
FROM events
GROUP BY table_name;

-- Примеры:
-- events: lag 5 minutes (отлично)
-- transactions: lag 6 hours (приемлемо)
-- revenue_summary: lag 25 hours (плохо, должна быть в 23:00)

SLA (Service Level Agreement) для разных данных:

• Real-time события (клики): < 5 минут
• Трансакционные (платежи): < 1 часа
• Отчёты (revenue): < 1 дня
• Аналитика (когорты): < 1 недели

Как улучшить: добавить мониторинг задержек, настроить алерты.

6. Validity (Валидность) — в правильном ли формате данные?

Данные соответствуют определённому схеме и формату.

# Проверка типов и форматов
import pandas as pd
from datetime import datetime

data = pd.read_sql("SELECT user_id, email, registration_date FROM users")

# Валидность: user_id должен быть UUID
invalid_user_ids = data[~data['user_id'].str.match(r'^[a-f0-9]{8}-[a-f0-9]{4}', na=False)]
validity_user_id = 100 * (1 - len(invalid_user_ids) / len(data))

# Валидность: email
invalid_emails = data[~data['email'].str.contains(r'^[\w\.-]+@[\w\.-]+\.\w+$', na=False)]
validity_email = 100 * (1 - len(invalid_emails) / len(data))

print(f"User ID validity: {validity_user_id:.1f}%")
print(f"Email validity: {validity_email:.1f}%")

Примеры ошибок валидности:

• phone в формате "+1-800-123-4567" вместо "+18001234567" ❌
• date как строка "2024-01-01" вместо DATE типа ❌
• JSON вместо структурированного столбца ❌

Как улучшить: использовать типизацию в БД и валидацию на уровне приложения.

Как создать систему мониторинга качества данных

Вариант 1: SQL dashboard (простой)

-- Таблица для отслеживания метрик качества
CREATE TABLE data_quality_metrics (
    id SERIAL PRIMARY KEY,
    check_date DATE,
    table_name VARCHAR(255),
    metric_name VARCHAR(100),  -- completeness, accuracy, etc
    metric_value FLOAT,
    threshold_warning FLOAT,
    threshold_critical FLOAT,
    status VARCHAR(20),  -- OK, WARNING, CRITICAL
    created_at TIMESTAMP DEFAULT NOW()
);

-- Ежедневный job для расчёта метрик
INSERT INTO data_quality_metrics 
SELECT 
    CURRENT_DATE,
    'users',
    'email_completeness',
    100.0 * COUNT(CASE WHEN email IS NOT NULL THEN 1 END) / COUNT(*),
    95.0,  -- warning при < 95%
    85.0,  -- critical при < 85%
    CASE 
        WHEN 100.0 * COUNT(CASE WHEN email IS NOT NULL THEN 1 END) / COUNT(*) >= 95 THEN 'OK'
        WHEN 100.0 * COUNT(CASE WHEN email IS NOT NULL THEN 1 END) / COUNT(*) >= 85 THEN 'WARNING'
        ELSE 'CRITICAL'
    END,
    NOW()
FROM users;

Вариант 2: Great Expectations (Python, рекомендуется)

import great_expectations as gx

context = gx.get_context()
validator = context.get_validator(
    batch_request={
        "datasource_name": "my_postgres",
        "data_connector_name": "default_postgres_data_connector",
        "data_asset_name": "users",
    }
)

# Определяем expectations (ожидания)
validator.expect_column_values_to_not_be_null("email")
validator.expect_column_value_lengths_to_be_between("email", min_value=5, max_value=255)
validator.expect_column_values_to_match_regex("email", regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
validator.expect_column_values_to_be_in_set("status", value_set=["active", "inactive", "pending"])
validator.expect_column_values_to_be_between("age", min_value=0, max_value=150)

# Запускаем валидацию
validation_result = validator.validate()
validation_result.save_as_expectation_suite(expectation_suite_name="users_suite")

# Report
print(validation_result.to_json_dict())

Вариант 3: dbt tests (для data warehouse)

# dbt/models/staging/stg_users.yml
models:
  - name: stg_users
    columns:
      - name: user_id
        tests:
          - unique
          - not_null
          - dbt_utils.expression_is_true:
              expression: "user_id ~ '^[a-f0-9]{8}'"
      
      - name: email
        tests:
          - not_null
          - unique
          - matches_regex:
              regex: "^[\w\.-]+@[\w\.-]+\.\w+$"
      
      - name: created_at
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: "created_at <= NOW()"

Внедрение Data Quality Framework

Шаг 1: Идентифицировать критичные данные

# Какие таблицы и колонки критичны для бизнеса?
critical_tables = {
    'users': {'importance': 'HIGH', 'sla_hours': 1},
    'transactions': {'importance': 'CRITICAL', 'sla_hours': 0.5},
    'events': {'importance': 'HIGH', 'sla_hours': 0.25},
    'marketing_campaigns': {'importance': 'MEDIUM', 'sla_hours': 24},
}

critical_columns = {
    'users': ['user_id', 'email', 'subscription_status'],
    'transactions': ['transaction_id', 'user_id', 'amount', 'timestamp'],
}

Шаг 2: Установить пороги качества

quality_thresholds = {
    'completeness': {
        'CRITICAL': 0.99,  # 99% непустых
        'HIGH': 0.95,
        'MEDIUM': 0.90,
    },
    'uniqueness': {
        'CRITICAL': 1.0,
        'HIGH': 0.99,
    },
    'accuracy': {
        'CRITICAL': 0.99,
        'HIGH': 0.95,
    },
}

Шаг 3: Автоматизировать проверки

# Ежечасный job
0 * * * * python /scripts/data_quality_check.py

# Результаты отправляются в Slack
if quality_score < threshold:
    send_alert_to_slack(f"Data quality issue in {table_name}: {metric}")

Шаг 4: Создать дашборд для отслеживания

-- Дашборд: Data Quality Status
SELECT 
    table_name,
    AVG(metric_value) as avg_quality_score,
    COUNT(CASE WHEN status = 'CRITICAL' THEN 1 END) as critical_issues,
    COUNT(CASE WHEN status = 'WARNING' THEN 1 END) as warning_issues,
    CASE 
        WHEN AVG(metric_value) >= 95 THEN '✅ Healthy'
        WHEN AVG(metric_value) >= 85 THEN '⚠️ Degraded'
        ELSE '❌ Critical'
    END as health_status
FROM data_quality_metrics
WHERE check_date >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY table_name
ORDER BY avg_quality_score DESC;

Частые ошибки при работе с качеством данных

❌ Ошибка 1: Игнорировать качество до крупного инцидента

"Всё работает, не трогаем" → потом случается catastrophic failure и все потеряно.

✅ Решение: проактивно мониторить от дня 1.

❌ Ошибка 2: Проверять только одно измерение (например, completeness)

Данные полные (100%) но некорректные (accuracy 40%).

✅ Решение: проверять все 6 измерений.

❌ Ошибка 3: Не иметь root cause анализа

"Email completeness упала на 10%" — но почему?

✅ Решение: для каждого инцидента найти причину и исправить её, не только симптом.

❌ Ошибка 4: Слишком строгие пороги

Требовать 100% completeness нереально. Пороги должны быть реалистичны.

✅ Решение: установить пороги на основе истории и SLA.

Случай из жизни: как плохие данные стоили $1m

Проблема: revenue report завышал на 12%

Причина: duplicate transactions не дедуплицировались
В таблице transactions было 200k записей,
но 24k из них были дубликаты (uniqueness = 88%)

Последствие: 
- Неправильные решения по budget allocation
- Overinvestment в неправильные каналы
- Потеря $1.2m из-за неправильной стратегии

Что помогло:
1. Внедрили Great Expectations на все таблицы
2. Добавили UNIQUE constraints в БД
3. Создали daily quality report в Slack
4. Теперь любой infodeficit détecté в течение часа

Итог

Data Quality — это не один-за-разом проект, а постоянный процесс. Главное правило: garbage in, garbage out. Вложение 10% времени в качество данных сэкономит 90% времени на debugging и переделку анализов.

Демо-контрольный список:

• ✅ Identify критичные данные
• ✅ Выбрать инструмент мониторинга (Great Expectations, dbt, custom)
• ✅ Установить пороги качества по 6 измерениям
• ✅ Автоматизировать проверки
• ✅ Создать дашборд и алерты
• ✅ Документировать все issues и их resolutions