0x03. dbt Tests & Sources - 데이터 품질을 코드로 보장하는 법

왜 테스트인가

데이터 파이프라인에서 가장 위험한 것은 조용한 실패(Silent Failure)이다. 쿼리는 성공했는데 결과가 틀린 경우. 중복 행이 섞여 있거나, NULL이 예상치 못한 곳에 들어가 있거나, 참조 무결성이 깨져 있는 경우. 이런 문제는 BI 리포트가 이상한 숫자를 보여줄 때까지 아무도 모른다.

dbt는 모델 빌드 직후 자동으로 테스트를 실행한다. 테스트 실패 시 다운스트림 모델의 실행을 중단하여 불량 데이터의 전파를 막는다.

Generic Tests (내장 4종)

YAML에서 선언적으로 정의한다. SQL을 직접 작성할 필요 없이 컬럼에 규칙을 붙인다.

# models/staging/_stg_jaffle_shop__models.yml
models:
  - name: stg_orders
    description: "주문 데이터 Staging 모델"
    columns:
      - name: order_id
        description: "주문 고유 ID"
        tests:
          - unique
          - not_null
      - name: status
        description: "주문 상태"
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned']
      - name: customer_id
        description: "고객 ID (FK)"
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id

테스트	검증 내용	실패 조건
`unique`	컬럼 값이 고유한가	중복 값 존재
`not_null`	NULL이 없는가	NULL 존재
`accepted_values`	허용된 값만 있는가	목록 외 값 존재
`relationships`	참조 무결성이 지켜지는가	FK가 PK에 없음

이 4종만으로도 대부분의 기본 품질 검증을 커버한다. 내부적으로 각 테스트는 결과가 0행이면 통과, 1행 이상이면 실패하는 SQL로 컴파일된다.

테스트 심각도 설정

- name: amount
  tests:
    - not_null:
        config:
          severity: warn  # 실패해도 파이프라인 중단 안 함
    - dbt_utils.accepted_range:
        min_value: 0
        config:
          severity: error  # 실패 시 파이프라인 중단

error (기본): 테스트 실패 시 파이프라인 중단
warn: 경고만 출력하고 계속 진행

프로덕션 초기에는 warn으로 시작해서 안정화되면 error로 전환하는 패턴이 일반적이다.

Singular Tests (커스텀)

내장 4종으로 표현할 수 없는 비즈니스 규칙은 SQL 파일로 직접 작성한다.

-- tests/assert_positive_amount.sql
SELECT order_id, amount
FROM {{ ref('stg_orders') }}
WHERE amount < 0

-- tests/assert_order_date_not_future.sql
SELECT order_id, order_date
FROM {{ ref('stg_orders') }}
WHERE order_date > CURRENT_DATE()

-- tests/assert_total_matches.sql
-- 주문 금액 합계와 결제 금액 합계가 일치하는지 검증
SELECT
    o.order_id,
    o.amount AS order_amount,
    p.total_paid
FROM {{ ref('fct_orders') }} o
JOIN (
    SELECT order_id, SUM(amount) AS total_paid
    FROM {{ ref('stg_payments') }}
    GROUP BY order_id
) p ON o.order_id = p.order_id
WHERE ABS(o.amount - p.total_paid) > 0.01

결과가 0행이면 통과, 1행 이상이면 실패. "금액은 양수여야 한다", "미래 날짜 주문은 없어야 한다", "주문-결제 금액이 일치해야 한다" 같은 비즈니스 규칙을 검증한다.

dbt build — 빌드와 테스트의 통합

dbt build는 모델 빌드와 테스트를 DAG 순서대로 인터리빙하여 실행한다.

stg_orders (빌드) → stg_orders 테스트 → fct_orders (빌드) → fct_orders 테스트

stg_orders의 테스트가 실패하면 fct_orders는 실행되지 않는다. 불량 데이터가 다운스트림으로 전파되는 것을 원천 차단한다.

dbt test로 테스트만 따로 실행할 수도 있다:

dbt test                        # 전체 테스트
dbt test -s stg_orders          # 특정 모델 테스트만
dbt test -s tag:critical        # 태그 기반 실행

dbt-utils와 dbt-expectations

서드파티 패키지로 테스트를 확장할 수 있다.

dbt-utils

- name: email
  tests:
    - dbt_utils.not_empty_string
    - dbt_utils.at_least_one
- name: created_at
  tests:
    - dbt_utils.recency:
        datepart: day
        field: created_at
        interval: 1  # 1일 이내 데이터 존재 확인

dbt-expectations (Great Expectations 스타일)

- name: amount
  tests:
    - dbt_expectations.expect_column_values_to_be_between:
        min_value: 0
        max_value: 100000
    - dbt_expectations.expect_column_mean_to_be_between:
        min_value: 10
        max_value: 1000

통계적 테스트(평균, 분포, 비율)까지 선언적으로 작성할 수 있다.

Sources — 외부 테이블 선언

웨어하우스에 이미 존재하는 테이블을 dbt 프로젝트에서 참조하려면 Source로 선언한다.

# models/staging/_sources.yml
sources:
  - name: jaffle_shop
    database: raw
    schema: jaffle_shop
    description: "Jaffle Shop 앱의 원본 데이터"
    tables:
      - name: orders
        description: "주문 테이블"
        loaded_at_field: _etl_loaded_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}
        columns:
          - name: id
            description: "주문 ID (PK)"
            tests:
              - unique
              - not_null
      - name: customers
        description: "고객 테이블"

Source Freshness — 데이터 신선도 모니터링

dbt source freshness

loaded_at_field로 지정한 컬럼의 최신 값을 확인한다. EL 단계에서 문제가 생겨 데이터가 늦게 도착하면 즉시 감지할 수 있다.

warn_after: 12 hours → 12시간 이상 새 데이터 없으면 경고
error_after: 24 hours → 24시간 이상이면 에러

CI/CD 파이프라인에서 dbt source freshness를 빌드 전에 실행하면, 오래된 소스 데이터로 모델을 빌드하는 실수를 방지할 수 있다.

Seeds — CSV를 테이블로

Seeds는 프로젝트 내 CSV 파일을 웨어하우스 테이블로 적재하는 기능이다.

-- seeds/country_codes.csv
code,name,region
KR,South Korea,Asia
US,United States,North America
JP,Japan,Asia

dbt seed  # CSV를 테이블로 적재

-- 모델에서 참조
SELECT o.*, c.region
FROM {{ ref('stg_orders') }} o
JOIN {{ ref('country_codes') }} c ON o.country_code = c.code

사용처: 국가 코드, 상태 코드, 부서 매핑 같은 소규모 정적 매핑 데이터에만 사용한다. 대용량이나 자주 변경되는 데이터는 EL 도구로 적재하고 Source로 선언해야 한다.