Что такое Data Contracts и какую проблему они решают?

Что такое Data Contracts и какую проблему они решают?

Data Contracts — это явное соглашение (контракт) между producer'ом (например, backend API или data pipeline) и consumer'ом (аналитик, ML модель, другой сервис), которое определяет структуру, формат и качество данных.

Это очень молодая, но критичная концепция в современном Data Engineering.

Проблема, которую решают Data Contracts

Сценарий: без контрактов

Backend team:
- Добавляет новый field "is_premium" к User API
- Удаляет поле "legacy_id" без предупреждения
- Изменяет формат "birth_date" с YYYY-MM-DD на unix timestamp

Data team (через час):
- ETL падает: неожиданный field "is_premium"
- Analytics падает: "legacy_id" используется в сегментации
- ML модель деградирует: "birth_date" теперь число вместо строки

Результат:
- 3 часа downtime
- Потеряны данные
- Срочные совещания и blame game

С Data Contracts:

Backend team:
- Хочет добавить "is_premium"
- Хочет удалить "legacy_id"
- Хочет изменить формат "birth_date"

Data contract проверяет:
- ✗ is_premium: не в контракте
- ✗ legacy_id удаляется: есть активные consumers
- ✗ birth_date: изменение типа нарушает контракт

Backend team получает ошибку перед деплоем.
Они уведомляют data team и согласовывают изменение.

Структура Data Contract

Пример для Order table/API

name: orders
version: 1.0.0
description: "Orders from e-commerce platform"

owner: payments-team
contact: payments-oncall@company.com

changelog:
  - version: 1.0.0
    date: 2024-01-01
    changes: "Initial version"

schema:
  fields:
    - name: order_id
      type: UUID
      nullable: false
      description: "Unique order identifier"
      required: true  # Consumer MUST use this field
    
    - name: user_id
      type: UUID
      nullable: false
      description: "User who placed the order"
      required: true
    
    - name: amount
      type: DECIMAL(10, 2)
      nullable: false
      description: "Order total in USD"
      required: true
    
    - name: status
      type: ENUM
      enum_values: ["pending", "confirmed", "shipped", "delivered", "cancelled"]
      nullable: false
      required: true
    
    - name: created_at
      type: TIMESTAMP
      nullable: false
      description: "Order creation time (UTC)"
      required: true
    
    - name: metadata
      type: JSON
      nullable: true
      description: "Additional order data"
      required: false  # Consumer SHOULD handle missing field

sla:  # Service Level Agreement
  latency:
    p50: 100ms
    p95: 500ms
    p99: 2000ms
  
  freshness:
    max_age: 15 minutes  # Data не должна быть старше 15 минут
  
  completeness:
    min_percentage: 99.9  # Минимум 99.9% полных records
  
  availability:
    uptime: 99.99%

owners:
  upstream:  # Кто создает эти данные
    - order-service
  downstream:  # Кто использует
    - analytics-warehouse
    - ml-fraud-detection
    - reporting-dashboard

changes_policy:
  breaking_changes_require:
    - approval_from: analytics-team
    - deprecation_period: 30 days
    - notification: all_downstream_teams

Типы изменений

Non-breaking changes (безопасно)

- Добавить новый опциональный field
- Переименовать internal field (если API интерфейс не меняется)
- Расширить enum с новым значением (если обрабатывается как string)
- Улучшить документацию

Breaking changes (требует координации)

- Удалить существующий field
- Изменить type поля (int -> string)
- Изменить enum значения
- Изменить обязательность поля (required -> optional)
- Изменить SLA (увеличить latency, снизить freshness)

Как валидировать Data Contracts

1. Schema validation (при инсерте)

import json
from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "properties": {
        "order_id": {"type": "string", "format": "uuid"},
        "user_id": {"type": "string", "format": "uuid"},
        "amount": {"type": "number", "minimum": 0},
        "status": {"enum": ["pending", "confirmed", "shipped", "delivered", "cancelled"]},
        "created_at": {"type": "string", "format": "date-time"}
    },
    "required": ["order_id", "user_id", "amount", "status", "created_at"]
}

def validate_order(order):
    try:
        validate(instance=order, schema=schema)
        return True, None
    except ValidationError as e:
        return False, str(e)

# Использование
order = {"order_id": "550e8400...", "user_id": "...", ...}
valid, error = validate_order(order)
if not valid:
    raise ValueError(f"Order validation failed: {error}")

2. SLA monitoring (в production)

import time
from prometheus_client import Histogram, Gauge

# Метрики
latency = Histogram('order_latency_ms', 'Order processing latency')
freshness = Gauge('order_freshness_minutes', 'Minutes since last update')
completeness = Gauge('order_completeness_percent', 'Percentage of complete records')

def monitor_sla(orders):
    # Latency
    start = time.time()
    # ... process orders
    latency.observe((time.time() - start) * 1000)
    
    # Freshness
    max_age_minutes = (datetime.now(UTC) - max(o['created_at'] for o in orders)).total_seconds() / 60
    freshness.set(max_age_minutes)
    
    if max_age_minutes > 15:
        alert("SLA violation: freshness")
    
    # Completeness
    complete_count = sum(1 for o in orders if all(o.get(f) for f in ['order_id', 'user_id', 'amount']))
    completeness_pct = (complete_count / len(orders)) * 100
    completeness.set(completeness_pct)
    
    if completeness_pct < 99.9:
        alert("SLA violation: completeness")

3. Integration tests

import pytest
from contracts import load_contract, validate_schema

class TestOrdersDataContract:
    
    @pytest.fixture
    def contract(self):
        return load_contract('orders.yaml')
    
    def test_schema_validation(self, contract):
        # Валидный order
        valid_order = {
            "order_id": "550e8400-e29b-41d4-a716-446655440000",
            "user_id": "user-123",
            "amount": 99.99,
            "status": "confirmed",
            "created_at": "2024-03-26T10:00:00Z"
        }
        assert contract.validate(valid_order)
        
        # Невалидный order
        invalid_order = {
            "order_id": "not-uuid",
            "user_id": "user-123",
            "amount": -10,  # Не может быть отрицательным
            "status": "unknown",  # Не в enum
            "created_at": "invalid-date"
        }
        assert not contract.validate(invalid_order)
    
    def test_required_fields(self, contract):
        # Все required fields должны быть present
        order_without_amount = {
            "order_id": "550e8400...",
            "user_id": "user-123",
            "status": "confirmed",
            "created_at": "2024-03-26T10:00:00Z"
        }
        assert not contract.validate(order_without_amount)
    
    def test_optional_fields(self, contract):
        # Optional fields могут быть absent
        order_without_metadata = {
            "order_id": "550e8400...",
            "user_id": "user-123",
            "amount": 99.99,
            "status": "confirmed",
            "created_at": "2024-03-26T10:00:00Z"
            # metadata пропущен, но это OK
        }
        assert contract.validate(order_without_metadata)

Инструменты для Data Contracts

1. Great Expectations

import great_expectations as ge

df = ge.read_csv('orders.csv')

expectations = {
    'order_id': {'type': 'string', 'unique': True},
    'amount': {'min': 0, 'max': 1000000},
    'status': {'values': ['pending', 'confirmed', 'shipped']}
}

for column, rules in expectations.items():
    for rule, value in rules.items():
        if rule == 'type':
            df.expect_column_values_to_be_of_type(column, value)
        elif rule == 'unique':
            df.expect_column_values_to_be_unique(column)
        elif rule == 'min':
            df.expect_column_values_to_be_between(column, min_value=value)
        elif rule == 'values':
            df.expect_column_values_to_be_in_set(column, value_set=value)

validation_result = df.validate()
if not validation_result.success:
    raise ValueError(f"Data contract violation: {validation_result.failures}")

2. Databand (monitoring)

from databand import task

@task
def process_orders(orders_df):
    # Автоматически мониторит:
    # - Schema changes
    # - Data quality metrics
    # - Performance
    return orders_df.filter(orders_df.amount > 0)

3. dbt + custom tests

-- models/staging/stg_orders.sql
{{ config(
    contract={
        enforced: true,
        checksum: "..."  # Меняется если schema меняется
    }
) }}

select
    order_id,
    user_id,
    amount,
    status,
    created_at
from {{ source('raw', 'orders') }}

-- tests/test_orders_contract.sql
select *
from {{ ref('stg_orders') }}
where
    amount < 0 or
    status not in ('pending', 'confirmed', 'shipped', 'delivered', 'cancelled') or
    created_at > current_timestamp()

Best Practices

1. Версионируй контракты

   • Семантическое версионирование (major.minor.patch)
   • major = breaking change требует миграции
   • minor = backward compatible addition

2. Документируй изменения

   • Всегда добавляй changelog
   • Объясняй WHY, не только WHAT

3. Управляй жизненным циклом

   • Proposed -> Accepted -> Deprecated -> Sunset
   • Минимум 30 дней между changes для migration

4. Уведомляй downstream consumers

   • Automated notifications при изменении
   • Дай time для адаптации

5. Автоматизируй валидацию

   • Checks в CI/CD pipeline
   • Prevent deployment если contract violated

Выводы

Data Contracts решают критичную проблему:

• Для producers: понимают, кто использует их данные и как
• Для consumers: получают гарантии о структуре и качестве
• Для organization: снижают downtime, улучшают quality, уменьшают friction между teams

Это особенно важно в крупных organizations с множеством data teams и сложными dependencies. Data contracts становятся инфраструктурой для reliable data-driven systems, как API contracts в микросервисах.