3. dbt 구성요소 살펴보기

-- models/staging/stg_orders.sql
-- 목적:
--   원본 주문 데이터(raw.orders)를 받아
--   컬럼명·타입을 표준화하고
--   향후 조인·집계가 쉽게 정리된 결과로 만든다.

{{ config(
  materialized = 'view'  -- 결과를 뷰로 생성
) }}

select
    order_id,                              -- 주문 ID (원본 그대로 사용)
    customer_id,                           -- 고객 ID
    cast(order_ts as timestamp) as order_time  -- 문자열/숫자 → timestamp 변환
from {{ source('raw', 'orders') }}         -- 원본 테이블 참조 (하드코딩 방지)

# models/sources.yml
version: 2

sources:
  - name: raw                    # 원본 스키마 이름 (논리적)
    description: "원본 시스템에서 추출된 데이터 레이어"
    tables:
      - name: orders
        description: "주문 거래 로그 원본 데이터"
        columns:
          - name: order_id
            description: "주문 ID, 고유 식별자"
            tests:                # 데이터 품질 테스트
              - not_null
              - unique
      - name: customers
        description: "고객 기본 정보"

-- models/staging/stg_orders.sql
{{ config(materialized='view') }}

select
    order_id,
    customer_id,
    cast(order_ts as timestamp) as order_time
from {{ source('raw', 'orders') }}   -- 원본 데이터 참조 (YAML 정의 기반)

# data/country_codes.csv
country_code,country_name
KR,South Korea
US,United States
JP,Japan

# seeds 로드
dbt seed

-- models/marts/country_summary.sql
{{ config(materialized='table') }}

select
    cc.country_name,
    count(o.order_id) as total_orders
from {{ ref('country_codes') }} as cc       -- CSV를 테이블처럼 참조
left join {{ ref('orders') }} as o
    on cc.country_code = o.country_code
group by cc.country_name

-- snapshots/customers_snapshot.sql

{% snapshot customers_snapshot %}
    {{
        config(
            target_schema='snapshots',      -- 이력 테이블 저장 위치
            unique_key='customer_id',       -- 변경 여부 판별 기준
            strategy='check',               -- 'check' 또는 'timestamp'
            check_cols=['email', 'address'] -- 변경 감지할 컬럼
        )
    }}

    select
        customer_id,
        email,
        address,
        updated_at
    from {{ source('raw', 'customers') }}

{% endsnapshot %}

# Snapshot 실행
dbt snapshot

-- models/stg_orders.sql
{{ config(materialized='view') }}
select
    order_id,
    customer_id,
    cast(order_ts as timestamp) as order_time
from {{ source('raw', 'orders') }}

-- models/fct_orders.sql
{{ config(materialized='table') }}
select
    order_id,
    customer_id,
    sum(amount) as total_amount
from {{ ref('stg_orders') }}
group by order_id, customer_id

-- models/inc_orders.sql
{{ config(
    materialized='incremental',
    unique_key='order_id'
) }}
select
    order_id,
    customer_id,
    sum(amount) as total_amount
from {{ source('raw', 'orders') }}
{% if is_incremental() %}
  where order_ts > (select max(order_time) from {{ this }})
{% endif %}
group by order_id, customer_id

# models/core/customers.yml
version: 2 # dbt YAML 메타데이터 스키마 버전

models: # 테스트를 적용할 모델 목록
  - name: customers
    description: "고객 정보 테이블"
    columns:
      - name: customer_id
        description: "고유 고객 식별자"
        tests: #  dbt 내장 테스트 키워드 (별도 SQL 작성 없이 사용 가능)
          - not_null       # 모든 customer_id가 반드시 존재해야 함
          - unique         # 중복된 customer_id가 없어야 함

-- tests/positive_order_amount.sql
select *
from {{ ref('orders') }} # 테스트 대상 모델 참조
where amount <= 0 # 주문 금액이 0 이하인 경우를 실패 조건으로 지정

#쿼리 결과가 한 행이라도 있으면 테스트 실패로 처리

# models/core/orders.yml
# 이 정보는 dbt docs generate 시 HTML 문서로 변환되어 검색 가능
version: 2

models:
  - name: orders
	  # description -> 모델 또는 컬럼의 의미, 가공 방식, 주의사항 기록
    description: "고객 주문 내역 테이블. 원본 데이터는 raw.orders에서 가져와 가공."
    columns: # 각 컬럼별 설명을 개별적으로 추가
      - name: order_id
        description: "고유 주문 식별자"
      - name: customer_id
        description: "주문한 고객의 ID"
      - name: order_date
        description: "주문이 생성된 날짜"
      - name: total_amount
        description: "주문 총 금액(할인, 세금 포함)"

# models/staging/stg_sources.yml
# 원본 테이블의 역할과 생성 주기 등도 함께 기재 가능
version: 2

sources: # 외부/원본 데이터에 대한 설명을 추가
  - name: raw
    description: "원본 데이터베이스 스키마"
    tables:
      - name: orders
        description: "주문 원본 테이블 (운영 DB에서 추출)"

dbt docs generate   # 문서 파일 생성
dbt docs serve      # 로컬 웹 서버로 문서 확인

flowchart LR
    A[Sources<br/>원본 데이터] --> B[(Seeds<br/>CSV 참조 데이터)]
    A --> C[Staging Models<br/>표준화 변환]
    B --> C
    C --> D[Core Models<br/>비즈니스 로직]
    D --> E[Mart Models<br/>최종 분석 테이블]
    A --> F[Snapshots<br/>변경 이력]
    
    classDef source fill:#E3F2FD,stroke:#1565C0,stroke-width:1px;
    classDef seed fill:#FFF3E0,stroke:#EF6C00,stroke-width:1px;
    classDef staging fill:#E8F5E9,stroke:#2E7D32,stroke-width:1px;
    classDef core fill:#F3E5F5,stroke:#6A1B9A,stroke-width:1px;
    classDef mart fill:#FCE4EC,stroke:#AD1457,stroke-width:1px;
    classDef snapshot fill:#FFFDE7,stroke:#FBC02D,stroke-width:1px;

    class A source;
    class B seed;
    class C staging;
    class D core;
    class E mart;
    class F snapshot;