데이터 웨어하우스가 필요한 이유 – OLTP 한계부터 BigQuery·Snowflake 실무 적용까지

-- 운영 DB에서 마케터가 실행하고 싶은 분석 쿼리 예시
SELECT
    p.category,
    DATE_TRUNC('month', o.created_at) AS month,
    COUNT(DISTINCT o.user_id)         AS unique_buyers,
    SUM(oi.quantity * oi.price)       AS revenue,
    AVG(u.age)                        AS avg_buyer_age
FROM orders o
JOIN order_items oi  ON o.id = oi.order_id
JOIN products p      ON oi.product_id = p.id
JOIN users u         ON o.user_id = u.id
WHERE o.created_at >= '2022-01-01'
GROUP BY p.category, month
ORDER BY month, revenue DESC;

-- 문제: orders 테이블 5,000만 건 풀스캔 → 실행 시간 수십 분
-- 결과: 동시 접속 사용자의 주문 API 응답 시간 급증 → 서비스 장애

[행 지향 저장 구조 – OLTP에 최적화]
디스크 블록:
| user_id | name   | email           | created_at | status | amount |
| 1001    | 김철수 | kim@example.com | 2024-01-01 | PAID   | 35000  |
| 1002    | 이영희 | lee@example.com | 2024-01-02 | PAID   | 12000  |

분석 쿼리 "금액 합계" → 모든 행을 읽으며 amount만 추출 → 비효율

[컬럼 지향 저장 구조 – DW/OLAP에 최적화]
amount 컬럼 블록: | 35000 | 12000 | 89000 | 45000 | ...
→ amount 컬럼 블록만 읽으면 즉시 합산 → 고효율

[스타 스키마 구조]

          dim_date
          (날짜 차원)
              │
dim_product ──┤── fact_sales ──┤── dim_customer
(상품 차원)   │   (판매 팩트)       (고객 차원)
              │
          dim_store
          (매장 차원)

fact_sales 컬럼:
  date_key       INT     → dim_date.date_key
  product_key    INT     → dim_product.product_key
  customer_key   INT     → dim_customer.customer_key
  store_key      INT     → dim_store.store_key
  quantity       INT     ← 측정값
  unit_price     DECIMAL ← 측정값
  discount_amt   DECIMAL ← 측정값
  total_revenue  DECIMAL ← 측정값

[ETL 전체 파이프라인]

┌─────────────────────────────────────────────────────────┐
│                      데이터 소스                          │
│  MySQL(주문DB) │ PostgreSQL(사용자DB) │ GA(웹로그) │ S3 │
└──────┬─────────────────┬────────────────┬──────────┬──┘
       │                 │                │          │
       └─────────────────┴────────────────┴──────────┘
                                │
                    ┌───────────▼───────────┐
                    │  E (Extract, 추출)     │
                    │  - 증분 추출(CDC)      │
                    │  - 전체 추출(Full)     │
                    │  - API 수집            │
                    └───────────┬───────────┘
                                │
                    ┌───────────▼───────────┐
                    │  T (Transform, 변환)   │
                    │  - 데이터 정제·정규화  │
                    │  - 타입 통일           │
                    │  - 비즈니스 룰 적용    │
                    │  - 중복 제거           │
                    └───────────┬───────────┘
                                │
                    ┌───────────▼───────────┐
                    │  L (Load, 적재)        │
                    │  - 스타 스키마로 변환  │
                    │  - 팩트/차원 테이블    │
                    │  - 파티션 최적화       │
                    └───────────┬───────────┘
                                │
                    ┌───────────▼───────────┐
                    │    데이터 웨어하우스    │
                    │  (BigQuery/Redshift)   │
                    └───────────────────────┘

# Apache Airflow DAG: MySQL → BigQuery ETL 파이프라인
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook
from datetime import datetime, timedelta
import pandas as pd

default_args = {
    'owner': 'data-engineering',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

def extract_orders(**context):
    """MySQL 운영 DB에서 전날 주문 데이터 증분 추출"""
    mysql_hook = MySqlHook(mysql_conn_id='prod_mysql')
    yesterday = context['ds']  # Airflow 실행 날짜

    query = f"""
        SELECT
            order_id, user_id, product_id,
            quantity, unit_price, status,
            created_at
        FROM orders
        WHERE DATE(created_at) = '{yesterday}'
          AND status IN ('PAID', 'DELIVERED')
    """
    df = mysql_hook.get_pandas_df(query)
    # XCom으로 다음 태스크에 전달 (소규모 데이터)
    return df.to_json()

def transform_orders(**context):
    """분석에 맞게 데이터 변환 및 정제"""
    raw_json = context['task_instance'].xcom_pull(task_ids='extract')
    df = pd.read_json(raw_json)

    # 변환 1: 총 금액 계산
    df['total_revenue'] = df['quantity'] * df['unit_price']

    # 변환 2: 날짜 차원 키 생성 (YYYYMMDD 형식)
    df['date_key'] = pd.to_datetime(df['created_at']).dt.strftime('%Y%m%d').astype(int)

    # 변환 3: 상태 코드 표준화 (운영 DB와 DW 간 코드 통일)
    status_map = {'PAID': 'COMPLETED', 'DELIVERED': 'COMPLETED'}
    df['status'] = df['status'].map(status_map)

    # 변환 4: 불필요 컬럼 제거
    df = df.drop(columns=['created_at'])

    return df.to_json()

def load_to_bigquery(**context):
    """변환된 데이터를 BigQuery fact_orders 테이블에 적재"""
    transformed_json = context['task_instance'].xcom_pull(task_ids='transform')
    df = pd.read_json(transformed_json)

    bq_hook = BigQueryHook(gcp_conn_id='google_cloud_default')
    bq_hook.insert_rows_from_dataframe(
        dataframe=df,
        project_id='my-project',
        dataset_id='dw_sales',
        table_id='fact_orders',
        chunk_size=10000
    )

with DAG(
    dag_id='orders_etl_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # 매일 새벽 2시 실행
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    extract = PythonOperator(task_id='extract', python_callable=extract_orders)
    transform = PythonOperator(task_id='transform', python_callable=transform_orders)
    load = PythonOperator(task_id='load', python_callable=load_to_bigquery)

    extract >> transform >> load  # 순서 정의

[ETL vs ELT 비교]

ETL (전통적 방식):
소스 → 변환(외부 서버) → DW 적재
장점: DW 저장 공간 절약
단점: 변환 서버 별도 필요, 재처리 어려움

ELT (현대적 방식):
소스 → DW 적재(Raw) → DW 내부에서 변환 (dbt 활용)
장점: 원본 데이터 보존, 재처리 용이, 변환 로직 버전 관리
단점: DW 스토리지 비용 증가 가능

[데이터 인프라 선택 가이드]

데이터가 주로 정형(테이블 형태)?
→ YES: 분석 목적이 SQL 쿼리·BI 대시보드 중심
         → 데이터 웨어하우스 선택
→ NO:  비정형(이미지, 로그, JSON 등) 또는 ML 학습 데이터
         → 데이터 레이크 선택

정형 + 비정형 동시에 필요하고
ML과 BI를 함께 사용?
         → 데이터 레이크하우스 선택
           (Databricks, Snowflake 통합 플랫폼)

-- BigQuery: 파티션과 클러스터링으로 쿼리 비용 최적화

-- 테이블 생성 (파티션 + 클러스터링)
CREATE TABLE `project.dw_sales.fact_orders`
(
    order_id      INT64,
    user_id       INT64,
    product_id    INT64,
    category      STRING,
    total_revenue FLOAT64,
    order_date    DATE
)
PARTITION BY order_date          -- 날짜로 파티셔닝 (스캔 범위 축소)
CLUSTER BY category, product_id; -- 자주 필터링하는 컬럼으로 클러스터링

-- 파티션 활용 쿼리: 특정 월 데이터만 스캔 (비용 절감)
SELECT
    category,
    SUM(total_revenue)  AS monthly_revenue,
    COUNT(DISTINCT user_id) AS unique_buyers
FROM `project.dw_sales.fact_orders`
WHERE order_date BETWEEN '2024-01-01' AND '2024-01-31'  -- 파티션 필터
  AND category = 'Electronics'                           -- 클러스터 필터
GROUP BY category
ORDER BY monthly_revenue DESC;

-- 비용 예측: 파티션 없으면 전체 테이블 스캔 → 파티션 사용 시 1/365 스캔

[DW 도입 필요 신호 체크리스트]

✅ 분석 쿼리가 운영 DB 성능에 영향을 미치기 시작했다
✅ 여러 시스템의 데이터를 통합 분석할 필요가 생겼다
✅ 수개월~수년 치 이력 데이터 분석이 필요해졌다
✅ BI 도구(Tableau, Looker, Superset)를 도입할 계획이다
✅ 데이터 분석가·사이언티스트가 팀에 합류했다
✅ 데이터 기반 의사결정이 비즈니스의 핵심이 되었다

[DW가 아직 불필요한 신호]

⬜ 데이터 소스가 DB 1개뿐이다
⬜ 팀 전체가 10명 이하이고 분석 업무가 드물다
⬜ 실시간 운영 데이터만 필요하고 이력 분석은 없다