팀 간 Data Contract

Contract 가 해결하는 문제

Producer 팀이 schema 변경을 배포해. Consumer 팀 파이프라인이 새벽 6시에 깨져. Consumer 팀은 화가 나. Producer 팀은 자기 테이블을 누가 읽는지도 몰랐어. — 데이터를 만드는 팀이 둘 이상인 회사에서 제일 흔한 cross-team 데이터 장애가 이 네 문장이야. 처방이 data contract 고.

Contract 에 들어가는 것

Schema — column 이름, type, null 허용 여부, 허용 값.
Freshness — "오전 9시까지 도착, 전체 날의 99.5% 에서."
소유권 — 만드는 팀 이름과 on-call 채널.
Breaking-change 정책 — column 을 없애기 며칠 전에 알릴지, rename 마이그레이션은 어떤 절차로 할지.
Versioning — additive 와 breaking 을 가르는 기준, downstream consumer 의 opt-in 여부.

Contract 가 아닌 것

Contract 는 아무도 안 읽는 위키 페이지가 아니야. 코드가 강제하는 산출물이야 — producer 의 write 단계에 schema 검증, 자동화된 freshness 체크, 그리고 모든 PR 에서 현재 schema 를 선언된 contract 와 대조하는 CI 테스트. 문서 속의 contract 는 희망 사항이고, CI 속의 contract 는 내력벽이야.

Code

Producer repo 에 check-in 된 YAML contract·yaml

# contracts/orders_v1.yaml
name: orders
version: 1
owner:
  team: orders-eng
  oncall: '#orders-data'
  email: orders-eng@example.com
freshness:
  sla_hours: 24
  measured_by: max(ingested_at)
schema:
  - name: order_id
    type: string
    required: true
    unique: true
    pattern: '^O\d{6}$'
  - name: customer_id
    type: string
    required: true
  - name: amount_usd
    type: number
    required: true
    minimum: 0
  - name: status
    type: string
    enum: [pending, completed, cancelled]
breaking_change_policy:
  notice_days: 30
  migration_pattern: |
    옆에 새 column 추가, notice window 동안 둘 다 채움,
    window 끝나면 옛 column 제거.

Producer 의 실제 schema 와 선언된 contract 비교하는 CI 체크·python

import yaml
import pyarrow.parquet as pq

def check_contract(parquet_path: str, contract_path: str) -> list[str]:
    '''실제 Parquet schema 와 contract 비교. 위반 list 반환.'''
    contract = yaml.safe_load(open(contract_path))
    actual = pq.read_schema(parquet_path)
    actual_cols = {f.name: str(f.type) for f in actual}
    declared = {c['name']: c['type'] for c in contract['schema']}

    # Contract 의 타입은 JSON-schema 어휘라, Arrow 가 실제로 보고하는 이름으로
    # 매핑해야 해. 안 그러면 타입 검사 쪽이 조용히 통과해.
    TYPE_MAP = {
        'string':  {'string', 'large_string'},
        'number':  {'double', 'float', 'halffloat'},
        'integer': {'int8', 'int16', 'int32', 'int64'},
        'boolean': {'bool'},
    }

    issues = []
    for name in declared.keys() - actual_cols.keys():
        issues.append(f'필수 column 누락: {name}')
    for name in actual_cols.keys() - declared.keys():
        issues.append(f'예상치 못한 column: {name} (contract 에 추가하거나 제거)')
    for name in declared.keys() & actual_cols.keys():
        allowed = TYPE_MAP.get(declared[name])
        if allowed and actual_cols[name] not in allowed:
            issues.append(
                f'타입 drift: {name} — contract 는 {declared[name]!r}, '
                f'파일은 {actual_cols[name]!r}')
    return issues

Exercise

본인이 만들고 다른 팀이 쓰는 데이터 product 하나를 골라. 한 페이지짜리 YAML contract 초안을 써 봐: schema, freshness, 소유권, breaking-change 정책. 그리고 쓰는 팀에 리뷰를 요청해. Contract 를 적는 과정 자체가 문서보다 값질 때가 많아 — 양쪽 다 당연한 줄 알았지만 서로 달랐던 가정들이 그때 다 드러나거든.