Как обеспечить качество данных в пайплайне?

Обеспечение качества данных в пайплайнах

Data Quality — фундамент надёжной аналитики. Плохие данные приводят к неверным решениям.

Что такое качество данных?

Данные высокого качества имеют:

• Completeness — полноту (нет пропущенных значений)
• Accuracy — точность (данные корректны)
• Consistency — консистентность (единообразие)
• Timeliness — актуальность (свежие)
• Validity — валидность (соответствуют схеме)
• Uniqueness — уникальность (нет дубликатов)

1. Валидация данных

На уровне dbt (лучшая практика)

version: 2
models:
  - name: stg_users
    columns:
      - name: user_id
        tests:
          - not_null
          - unique
      - name: email
        tests:
          - not_null
          - unique
      - name: age
        tests:
          - dbt_utils.expression_is_true:
              expression: "age >= 0 AND age <= 150"
      - name: status
        tests:
          - accepted_values:
              values: ['active', 'inactive', 'banned']

На уровне SQL

SELECT 
    COUNT(*) as total_rows,
    COUNT(*) FILTER (WHERE user_id IS NULL) as null_user_ids,
    COUNT(*) FILTER (WHERE age < 0 OR age > 150) as invalid_ages,
    COUNT(DISTINCT user_id) as unique_users
FROM raw.users;

2. Обнаружение дубликатов

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

-- Удалить (оставить первый)
DELETE FROM users
WHERE user_id IN (
    SELECT user_id FROM (
        SELECT user_id, 
               ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at) as rn
        FROM users
    ) subq
    WHERE rn > 1
);

3. Great Expectations

from great_expectations import dataset

def validate_orders(df):
    ge_df = dataset.PandasDataset(df)
    
    # Проверки
    ge_df.expect_column_values_to_be_not_null('order_id')
    ge_df.expect_column_values_to_be_of_type('amount', 'float')
    ge_df.expect_column_values_to_be_between('amount', 0, 999999)
    
    validation_result = ge_df.validate()
    if not validation_result['success']:
        raise ValueError(f"Validation failed")
    
    return df

4. Обнаружение аномалий

import numpy as np

def detect_anomalies(df, column, sensitivity=3):
    mean = df[column].mean()
    std = df[column].std()
    
    lower_bound = mean - sensitivity * std
    upper_bound = mean + sensitivity * std
    
    anomalies = df[
        (df[column] < lower_bound) | (df[column] > upper_bound)
    ]
    return anomalies

5. Freshness checks

sources:
  - name: raw
    tables:
      - name: orders
        loaded_at_field: created_at
        freshness:
          warn_after:
            count: 2
            period: day
          error_after:
            count: 3
            period: day

6. Integration тесты

def test_orders_not_null_amount():
    df = get_orders()
    assert df['amount'].isnull().sum() == 0

def test_orders_amount_positive():
    df = get_orders()
    assert (df['amount'] > 0).all()

def test_no_duplicates():
    df = get_orders()
    assert df['order_id'].nunique() == len(df)

Чеклист: Data Quality Framework

• Валидация схемы при загрузке
• Проверка null значений
• Проверка уникальности ключей
• Проверка диапазонов значений
• Обнаружение дубликатов
• Мониторинг freshness
• dbt тесты (not null, unique, relationships)
• Great Expectations для сложных проверок
• Elementary для аномалий
• Алерты при нарушениях

Вывод

Качество данных — постоянный мониторинг. Инвестируй в:

1. dbt tests (легко и быстро)
2. Great Expectations (для сложных сценариев)
3. Мониторинг аномалий
4. Документирование SLA
5. Автоматизацию проверок

Плохие данные дороже, чем хорошие пайплайны.