3. Silver 레이어 - Bronze를 분석 가능한 상태로 올린다

!pip install -q duckdb dbt-core dbt-duckdb

import os

# dbt 프로젝트 디렉토리 구조 생성
os.makedirs('jaffle_shop/models/staging', exist_ok=True)
os.makedirs('jaffle_shop/models/marts', exist_ok=True)

%%writefile jaffle_shop/dbt_project.yml
name: 'jaffle_shop'
version: '1.0.0'
profile: 'jaffle_shop'

model-paths: ["models"]

%%writefile jaffle_shop/profiles.yml
jaffle_shop:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: /content/warehouse.duckdb

%%writefile jaffle_shop/models/staging/stg_orders.sql
with source as (
    select * from bronze.orders
),

cleaned as (
    select
        id as order_id,
        user_id as customer_id,
        cast(order_date as date) as order_date,
        status
    from source
)

select * from cleaned

%%writefile jaffle_shop/models/staging/stg_customers.sql
with source as (
    select * from bronze.customers
),

cleaned as (
    select
        id as customer_id,
        first_name,
        last_name
    from source
)

select * from cleaned

%%writefile jaffle_shop/models/staging/stg_payments.sql
with source as (
    select * from bronze.payments
),

cleaned as (
    select
        id as payment_id,
        order_id,
        payment_method,
        amount / 100.0 as amount_dollars
    from source
)

select * from cleaned

!cd jaffle_shop && dbt run --select staging.*

import duckdb

conn = duckdb.connect('warehouse.duckdb')

# Silver 레이어 확인
conn.execute("SELECT * FROM stg_orders LIMIT 5").fetchdf()

# 타입 확인 — order_date가 DATE로 바뀌었는가
conn.execute("""
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_name = 'stg_orders'
""").fetchdf()

# payments의 amount가 달러 단위로 변환되었는가
conn.execute("SELECT * FROM stg_payments LIMIT 5").fetchdf()

with source as (
    -- 1단계: Bronze에서 원본을 가져온다
    select * from bronze.orders
),

cleaned as (
    -- 2단계: 정제 로직을 적용한다
    select
        id as order_id,
        cast(order_date as date) as order_date
    from source
)

-- 3단계: 최종 결과를 반환한다
select * from cleaned

with source as (
    select * from bronze.orders
),

deduplicated as (
    select
        *,
        row_number() over (
            partition by id
            order by _loaded_at desc
        ) as row_num
    from source
),

cleaned as (
    select
        id as order_id,
        user_id as customer_id,
        cast(order_date as date) as order_date,
        status
    from deduplicated
    where row_num = 1
)

select * from cleaned

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id='silver_transformation',
    schedule='0 6 * * *',
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    # Bronze 적재 완료를 기다린 뒤 Silver 변환 실행
    run_staging = BashOperator(
        task_id='dbt_run_staging',
        bash_command='cd /opt/dbt/jaffle_shop && dbt run --select staging',
    )

    # dbt test로 Silver 데이터 품질 검증
    test_staging = BashOperator(
        task_id='dbt_test_staging',
        bash_command='cd /opt/dbt/jaffle_shop && dbt test --select staging',
    )

    run_staging >> test_staging