2. Bronze 레이어 - 원본을 있는 그대로 쌓는다

타임스탬프 기반:  WHERE updated_at > '마지막 적재 시점'
자동 증가 키:    WHERE order_id > 마지막_적재_ID
CDC:            데이터베이스 변경 로그 캡처

import duckdb

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

# 소스 데이터가 변경된 상황을 시뮬레이션
# 실제로는 소스 시스템에서 SELECT * 로 전체를 가져온다

conn.execute("""
-- Full Load: 통째로 교체
CREATE OR REPLACE TABLE bronze.orders AS
SELECT *
FROM read_csv_auto(
  'https://raw.githubusercontent.com/dbt-labs/jaffle_shop/main/seeds/raw_orders.csv'
);
""")

print("Full Load 완료:",
      conn.execute("SELECT count(*) FROM bronze.orders").fetchone()[0], "건")

# 워터마크 테이블: 마지막 적재 지점을 기록
conn.execute("""
CREATE TABLE IF NOT EXISTS bronze.watermarks (
    table_name VARCHAR PRIMARY KEY,
    last_loaded_id INTEGER,
    last_loaded_at TIMESTAMP DEFAULT current_timestamp
);
""")

# 현재 워터마크 확인
watermark = conn.execute("""
SELECT COALESCE(last_loaded_id, 0)
FROM bronze.watermarks
WHERE table_name = 'orders'
""").fetchone()

last_id = watermark[0] if watermark else 0
print(f"마지막 적재 ID: {last_id}")

# 증분 적재: last_id 이후 데이터만 가져온다
conn.execute(f"""
INSERT INTO bronze.orders
SELECT *
FROM read_csv_auto(
  'https://raw.githubusercontent.com/dbt-labs/jaffle_shop/main/seeds/raw_orders.csv'
)
WHERE id > {last_id};
""")

# 워터마크 갱신
conn.execute("""
INSERT OR REPLACE INTO bronze.watermarks (table_name, last_loaded_id, last_loaded_at)
SELECT 'orders', MAX(id), current_timestamp
FROM bronze.orders;
""")

print("Incremental Load 완료")

conn.execute("""
CREATE OR REPLACE TABLE bronze.orders_with_meta AS
SELECT
    *,
    current_timestamp AS _loaded_at,
    'jaffle_shop' AS _source_system,
    'full' AS _load_type
FROM read_csv_auto(
  'https://raw.githubusercontent.com/dbt-labs/jaffle_shop/main/seeds/raw_orders.csv'
);
""")

conn.execute("SELECT * FROM bronze.orders_with_meta LIMIT 3").fetchdf()

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import duckdb

def load_full(table_name, source_url, **context):
    """Full Load: 전체 교체"""
    conn = duckdb.connect('warehouse.duckdb')
    conn.execute(f"""
        CREATE OR REPLACE TABLE bronze.{table_name} AS
        SELECT *, current_timestamp AS _loaded_at,
               '{table_name}' AS _source_system, 'full' AS _load_type
        FROM read_csv_auto('{source_url}')
    """)
    conn.close()

def load_incremental(table_name, source_url, key_column, **context):
    """Incremental Load: 워터마크 이후만"""
    conn = duckdb.connect('warehouse.duckdb')
    wm = conn.execute(f"""
        SELECT COALESCE(last_loaded_id, 0)
        FROM bronze.watermarks WHERE table_name = '{table_name}'
    """).fetchone()
    last_id = wm[0] if wm else 0

    conn.execute(f"""
        INSERT INTO bronze.{table_name}
        SELECT *, current_timestamp AS _loaded_at
        FROM read_csv_auto('{source_url}')
        WHERE {key_column} > {last_id}
    """)
    conn.close()

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

    # 소규모 마스터 → Full Load
    load_customers = PythonOperator(
        task_id='load_customers_full',
        python_callable=load_full,
        op_kwargs={'table_name': 'customers', 'source_url': '...'},
    )

    # 대용량 트랜잭션 → Incremental Load
    load_orders = PythonOperator(
        task_id='load_orders_incremental',
        python_callable=load_incremental,
        op_kwargs={
            'table_name': 'orders',
            'source_url': '...',
            'key_column': 'id',
        },
    )

    # 병렬 실행 — 테이블 간 의존 관계가 없으니까
    [load_customers, load_orders]