seongju
PORTFOLIO
“He who has a why to live can bear almost any how.” — Nietzsche“He who has a why to live can bear almost any how.” — Nietzsche“He who has a why to live can bear almost any how.” — Nietzsche“He who has a why to live can bear almost any how.” — Nietzsche

2026년 9월 10일

[3편] dbt build로 "이 데이터를 믿어도 되는가"를 파이프라인에 넣기

2편에서 이어서

1편에서 로컬에 파이프라인을 만들었고, 2편에서 실행 계층을 스케줄러에서 Celery worker로 옮겼다. 파이프라인은 돈다. 그런데 2편까지의 mart_daily_trips를 일자순으로 열어보면 맨 앞이 이렇다.

trip_date   trip_count  avg_fare
2001-01-01           3    47.97
2002-12-31           2    50.85
2008-12-31           8    26.40
...
2023-01-01       76752    21.87

NYC TLC 데이터는 2009년부터인데 2001년, 2008년 행이 앞에 붙어 있다. 원본 parquet에 승차 시각이 깨진 행이 소수 섞여 있어서다. 요금이 음수인 행도 있다. 변환이 성공했다는 것과 결과를 믿어도 된다는 것은 다른 얘기다. 이번 편은 그 간극을 dbt로 메운다.

지금 dbt가 하는 일과 안 하는 일

1편의 dbt 프로젝트는 raw_trips를 stg_trips(view)로 정리하고 mart_daily_trips(table)로 집계하는 것까지만 한다. 테스트가 하나도 없다. dbt의 핵심은 "SQL 변환을 소프트웨어처럼 버전 관리하고 테스트한다"인데, 지금은 앞부분만 쓰고 있는 셈이다.

이번 편에서 추가하는 것:

  • stg_trips에 정상 범위 필터를 넣고, 걸러진 행은 따로 집계한다.
  • 컬럼 값에 대한 dbt test를 붙이되, "파이프라인을 멈춰야 할 문제"와 "기록만 할 문제"를 나눈다.
  • dbt run을 dbt build로 바꿔서, 변환과 테스트가 같은 실행 단위 안에서 돌게 한다.

잡음을 어디서 거를까 — 설계 결정

깨진 행을 처리하는 방법은 크게 셋이다.

  1. raw는 그대로 두고 dbt test로 플래그만 한다. 필터 로직이 없어서 간단하지만 mart에 잡음이 그대로 남는다. "테스트가 알려주긴 하는데 결과물은 안 고쳐진다."
  2. stg_trips_valid / stg_trips_quarantined 두 모델로 쪼갠다. 실무 데이터 품질 파이프라인에 가깝지만, dbt 한 편에서 다루기엔 모델이 너무 늘어난다.
  3. stg_trips에서 필터하고, 걸러진 행은 audit 모델로 집계한다.

3번을 택했다. mart는 깨끗해지고, 버려진 데이터도 "몇 건이, 왜" 인지 dropped_trips 테이블에 남는다. raw → staging → (mart / audit)로 갈라지는 모델 레이어링을 보여주기에도 적당하다.

구현

flowchart LR
    RAW[("raw_trips")] --> STG["stg_trips<br/>(정상 범위만)"]
    RAW --> DROP["dropped_trips<br/>(걸러진 행 + 사유)"]
    STG --> MART["mart_daily_trips"]
    STG --> T["dbt test<br/>error 실패 → 중단<br/>warn 실패 → 기록만"]
    DROP --> T

stg_trips에 승차 시각 필터를 건다. 경계는 "NYC TLC 데이터는 2009년부터"라는 사실에서 온다.

-- models/staging/stg_trips.sql
{% set service_start = "'2009-01-01'" %}
{% set future_cutoff = "'2025-01-01'" %}

select
    tpep_pickup_datetime  as pickup_at,
    tpep_dropoff_datetime as dropoff_at,
    payment_type,
    trip_distance,
    fare_amount,
    total_amount
from {{ source('raw', 'raw_trips') }}
where tpep_pickup_datetime is not null
  and tpep_pickup_datetime >= {{ service_start }}
  and tpep_pickup_datetime <  {{ future_cutoff }}

걸러진 행은 사유와 함께 audit 모델로 남긴다.

-- models/audit/dropped_trips.sql
select
    tpep_pickup_datetime as pickup_at,
    fare_amount,
    case
        when tpep_pickup_datetime is null            then 'null_pickup'
        when tpep_pickup_datetime <  {{ service_start }} then 'pickup_before_service_start'
        when tpep_pickup_datetime >= {{ future_cutoff }} then 'pickup_in_future'
    end as drop_reason
from {{ source('raw', 'raw_trips') }}
where tpep_pickup_datetime is null
   or tpep_pickup_datetime <  {{ service_start }}
   or tpep_pickup_datetime >= {{ future_cutoff }}

테스트 — 무엇이 멈춰야 할 문제인가

dbt에는 두 종류의 test가 있다. 일반(generic) test는 not_null, unique, accepted_values처럼 이름 붙은 규칙을 YAML에서 컬럼에 갖다 붙이는 것이고, singular test는 tests/ 폴더에 SQL 파일 하나로 직접 쓰는 것이다(행을 하나도 반환하지 않으면 통과).

그리고 test에는 severity가 있다. error면 dbt build가 실패하고(그 태스크가 죽고 하류가 멈춘다), warn이면 로그만 남기고 통과한다. 어떤 검사를 어느 쪽에 둘지가 이 편의 핵심 설계다.

error로 둔 것 — 이게 깨지면 우리 코드가 잘못된 것:

  • pickup_at이 not_null이고 2009-01-01 이상 2025-01-01 이하. stg_trips의 WHERE가 이미 보장하는 값이다. 이 테스트가 실패하면 데이터가 아니라 필터 로직이 깨진 것이므로 파이프라인을 세우는 게 맞다.

warn으로 둔 것 — 상류 데이터 품질 문제이지 우리가 멈출 일은 아닌 것:

  • fare_amount >= 0
  • payment_type이 1~6
  • dropped_trips 행 수가 임계치 미만
# models/staging/schema.yml  (전체 중 발췌)
version: 2
models:
  - name: stg_trips
    columns:
      - name: pickup_at
        data_tests:
          - not_null
          - dbt_utils.accepted_range:   # min/max 양끝 포함
              arguments:
                min_value: "'2009-01-01'"
                max_value: "'2025-01-01'"
      - name: fare_amount
        data_tests:
          - dbt_utils.accepted_range:
              arguments: { min_value: 0 }
              config:
                severity: warn
                store_failures: true
      - name: payment_type
        data_tests:
          - accepted_values:
              arguments:
                values: [1, 2, 3, 4, 5, 6]
                quote: false
              config: { severity: warn, store_failures: true }

dbt 1.10부터 일반 test 인자는 arguments: 아래에 넣어야 하고, 키도 tests: → data_tests:로 바뀌었다. 구버전 문법은 아직 동작하지만 deprecation 경고가 뜬다.

store_failures: true면 실패한 행이 main_dbt_test__audit 스키마의 테이블로 남는다(main은 DuckDB의 기본 스키마다). "몇 건이 왜 걸렸나"를 나중에 조회할 수 있다.

범위 검사보다 복잡한 규칙은 singular test로 만든다.

-- tests/dropped_trips_within_tolerance.sql
-- 행이 반환되면(= 걸러진 행이 임계 초과) 실패
{{ config(severity="warn") }}
select count(*) from {{ ref('dropped_trips') }}
having count(*) > 500

2편에서 남긴 숙제 하나 — NYC TLC가 2023-02부터 airport_fee 컬럼명을 Airport_fee로 바꾼 것 — 도 여기서 처리한다. 단, dbt의 일반 test는 행의 값을 검사하지 컬럼의 존재를 검사하지 않는다. 컬럼이 사라지거나 이름이 바뀌는 스키마 변화는 information_schema를 직접 봐야 잡힌다.

-- tests/assert_airport_fee_present.sql
select 'airport_fee column missing from raw_trips' as failure
where not exists (
    select 1 from information_schema.columns
    where lower(table_name) = 'raw_trips'
      and lower(column_name) = 'airport_fee'
)

지금 구조에서는 이 테스트가 늘 통과한다. Load가 컬럼을 이름이 아니라 위치로 매칭하기 때문에(2편 참고), raw_trips의 컬럼명은 첫 적재(2023-01, 소문자 airport_fee) 시점에 고정되고 그 뒤로 안 바뀐다. 그래도 걸어두는 이유는, 나중에 누가 적재를 이름 기준으로 바꾸거나 raw_trips를 다시 만들면 그때 조용히 깨질 자리이기 때문이다. 지금 비용이 0인 canary는 미리 심어둔다.

dbt run → dbt build

1편의 Transform 태스크는 dbt run이었다. dbt build로 바꾼다.

  transform = BashOperator(
-     task_id="dbt_run",
-     bash_command=f"cd {DBT_PROJECT_DIR} && DBT_PROFILES_DIR={DBT_PROJECT_DIR} dbt run",
+     task_id="dbt_build",
+     bash_command=f"cd {DBT_PROJECT_DIR} && DBT_PROFILES_DIR={DBT_PROJECT_DIR} dbt build",
  )

dbt build는 모델을 만들고, 각 모델이 만들어진 직후 그 모델에 걸린 test를 DAG 순서대로 실행한다. stg_trips가 만들어지면 바로 stg_trips의 test가 돌고, 그게 error로 깨지면 하류 mart_daily_trips는 아예 만들어지지 않는다. 변환과 검증이 따로 노는 두 태스크가 아니라 하나의 실행 단위가 된다.

dbt_utils 같은 패키지는 패키지 매니페스트(packages.yml)에 적고 dbt deps로 받는다. 이 프로젝트는 받은 결과(dbt_packages/)를 레포에 커밋해서, 컨테이너를 다시 빌드해도 네트워크 없이 동작하게 해뒀다.

돌려보면 — PASS=11 WARN=2 ERROR=0

2023-01 ~ 2023-04 네 달치로 dbt build를 돌린 결과.

PASS=11  WARN=2  ERROR=0  TOTAL=13

(TOTAL 13 = 모델 3 + test 10. 위에 나온 6개(YAML의 pickup_at 2·fare_amount 1·payment_type 1 + singular 2) + dropped_trips.drop_reason의 not_null/accepted_values 2개 + 1편부터 있던 mart_daily_trips.trip_date의 not_null/unique 2개 = 10개다.)

검사 종류 severity 왜 이번 실측
pickup_at not_null + 2009~2025 범위 일반 error stg 필터가 보장하는 값. 깨지면 필터 로직 버그 PASS
fare_amount >= 0 일반 warn 상류(NYC TLC) 품질. 음수 = 환불/조정 WARN 109,478행
payment_type 1~6 일반 warn 상류 품질 WARN 1건 (위반 값 0 하나)
drop_reason not_null + 값 3종 일반 error audit 모델 자체가 온전한지 PASS
dropped_trips 행 수 < 임계 singular warn 걸러진 양이 급증하면 소스 이상 신호 PASS (24행, 임계 500)
airport_fee 컬럼 존재 singular error 스키마 canary PASS
  • fare_amount 음수 109,478행은 버그가 아니다. NYC TLC는 환불·조정 거래를 음수 요금으로 기록한다. "알고 있어야 할 사실"이라 warn으로 둔다.
  • dropped_trips 24행은 전부 pickup_before_service_start다(null_pickup, pickup_in_future는 0건).
  • 두 warn의 실패 행은 main_dbt_test__audit 스키마 테이블에 그대로 남아 있어 나중에 들여다볼 수 있다.

error가 실제로 파이프라인을 멈추는지도 확인했다. dropped_trips_within_tolerance를 severity: error에 임계치 10으로 바꾸면(실제 24행) —

FAIL 1 dropped_trips_within_tolerance
Done. PASS=10 WARN=2 ERROR=1
$ echo $?
1

dbt build가 exit code 1로 끝나고, Airflow의 dbt_build 태스크가 실패한다. warn이었으면 exit 0, 태스크 성공.

정리하면

  • dbt는 raw를 mart로 바꾸는 SQL 실행기가 아니라, 그 SQL에 검증을 붙일 수 있게 하는 도구다. 1편에선 변환만 썼고, 이번에 test를 붙여야 절반이 마저 채워진다.
  • 잡음 행은 stg_trips에서 거르고 dropped_trips audit 모델로 남긴다. mart는 깨끗하게, 버린 데이터는 추적 가능하게.
  • test severity의 error / warn은 "파이프라인을 멈춰야 할 문제"와 "기록만 할 문제"를 나누는 장치다. 우리 필터가 보장하는 값은 error, 상류 데이터 품질은 warn.
  • dbt run → dbt build로 바꾸면 변환과 테스트가 DAG 순서대로 한 실행 단위 안에서 돈다. error 테스트가 깨지면 하류 모델은 만들어지지 않는다.

1편에서 파이프라인을 만들고, 2편에서 실행 계층을 확장하고, 3편에서 산출된 데이터가 스스로를 검증하게 했다. "클라우드 없이 여덟 개 기술로 ELT 한 바퀴"라는 1편의 목표는 이제 도는 파이프라인 + 갈아끼울 수 있는 실행 계층 + 자기 검증까지를 뜻한다. 남은 큰 조각은 warehouse다 — DuckDB 파일 하나를 서버형 DB로 바꾸면 동시성과 스키마 관리가 어떻게 달라지는지는 이 시리즈 밖의 이야기로 남겨둔다.

참고