ETL ELT 차이점 – 데이터 파이프라인 어떻게 선택할까

[현대 기업의 데이터 소스 현황]

트랜잭션 DB  ──→ MySQL, PostgreSQL (주문, 회원)
모바일 앱    ──→ Firebase, Amplitude (이벤트 로그)
웹 서버      ──→ Nginx 로그, Google Analytics
SaaS 도구    ──→ Salesforce, HubSpot, Stripe
IoT 센서     ──→ MQTT 스트림, 시계열 데이터
소셜 미디어  ──→ API (트위터, 인스타그램)
외부 데이터  ──→ 공공 API, 기상 데이터, 환율

문제:
  각 소스는 형식·스키마·갱신 주기가 모두 다름
  → 하나로 모아 정제하지 않으면 분석 불가능

해결:
  데이터 파이프라인 = 이 모든 소스를 통합하는 '데이터 고속도로'

E (Extract)   : 소스 시스템에서 원본 데이터 추출
T (Transform) : 데이터 정제·변환·집계·검증
L (Load)      : 목적지(DW, 데이터 레이크 등)에 적재

ETL: Extract → Transform → Load
     (목적지 도착 전에 변환 완료)

ELT: Extract → Load → Transform
     (목적지 도착 후 내부에서 변환)

[파이프라인 없는 조직의 고통]

데이터 사일로:  팀마다 다른 DB, 숫자가 제각각
               "우리 팀 매출은 50억인데 재무팀은 48억이라네요?"

분석 지연:     개발자에게 SQL 쿼리 요청 → 며칠 대기
수작업 오류:   엑셀 복붙 → 집계 실수 → 잘못된 의사결정
확장 불가:     데이터 10배 증가 시 기존 방식 전면 재작업

[ETL 전체 흐름]

소스 시스템들                ETL 엔진               목적지 DW
(MySQL, API, CSV)           (중간 처리 서버)        (Oracle, Teradata)

주문 DB  ─────────→ ┌─────────────────────┐
회원 DB  ─────────→ │   E: 데이터 추출      │
로그 파일 ────────→ │   (Staging Area 임시  │
외부 API ─────────→ │    저장)              │
                    │         ↓            │
                    │   T: 변환 처리        │
                    │   - 결측값 제거       │ ──→ ┌────────────┐
                    │   - 타입 변환         │     │  데이터 웨어 │
                    │   - 중복 제거         │     │  하우스(DW) │
                    │   - 비즈니스 규칙 적용│     │  (정제된    │
                    │   - 집계/조인         │     │   구조화    │
                    │         ↓            │     │   데이터만) │
                    │   L: 적재            │     └────────────┘
                    └─────────────────────┘

핵심: 목적지 DW에 도달하기 전에 변환 완료
      DW에는 항상 깨끗하고 구조화된 데이터만 저장

# ETL 패턴: 적재 전 개인정보 마스킹 처리
import hashlib
import re

def mask_pii(record: dict) -> dict:
    """
    ETL Transform 단계 – 개인정보 마스킹
    DW에는 원본 개인정보가 절대 저장되지 않음
    """
    masked = record.copy()

    # 이메일: 앞 3글자만 남기고 마스킹
    if 'email' in masked:
        email = masked['email']
        local, domain = email.split('@')
        masked['email'] = local[:3] + '***@' + domain

    # 전화번호: 중간 4자리 마스킹
    if 'phone' in masked:
        phone = re.sub(r'(\d{3})-(\d{4})-(\d{4})', r'\1-****-\3', masked['phone'])
        masked['phone'] = phone

    # 주민등록번호: 해시로 대체 (원본 삭제)
    if 'ssn' in masked:
        masked['ssn_hash'] = hashlib.sha256(masked['ssn'].encode()).hexdigest()
        del masked['ssn']   # 원본 완전 삭제

    return masked

# 사용 예시
raw_record = {
    'user_id': 12345,
    'email':   'honggildong@example.com',
    'phone':   '010-1234-5678',
    'ssn':     '900101-1234567',
    'amount':  50000
}

safe_record = mask_pii(raw_record)
print(safe_record)
# {
#   'user_id': 12345,
#   'email':   'hon***@example.com',
#   'phone':   '010-****-5678',
#   'ssn_hash': 'a3f9b2c1...',  ← 원본 없음
#   'amount':  50000
# }
# → 이 상태로만 DW에 적재

[ETL의 구조적 한계]

① 변환 로직 변경의 어려움
   과거 데이터에 새 비즈니스 규칙 적용 시
   → 전체 파이프라인 재실행 필요 (수 시간~수 일)
   "작년 매출에 새로운 환불 기준을 소급 적용하려면?"
   → ETL 파이프라인 전체 재처리

② 원본 데이터 소실
   변환 후 원본을 저장하지 않으면
   나중에 다른 방식으로 분석하고 싶어도 불가능
   "로그 원본이 있었으면 머신러닝 학습에 쓸 수 있었는데..."

③ 중간 처리 서버 병목
   대용량 데이터 처리 시 ETL 서버가 단일 병목
   소스 데이터 10TB → ETL 서버 처리 → DW 적재
   ETL 서버 메모리·CPU 한계가 파이프라인 처리량 결정

[ELT 전체 흐름]

소스 시스템들         경량 커넥터         클라우드 DW 내부
                     (Fivetran, Airbyte)

주문 DB  ──────────→ ┌───────────┐
회원 DB  ──────────→ │ E: 추출   │ ──→ ┌─────────────────────────┐
로그 파일 ─────────→ │           │     │  Raw Layer (원본 그대로) │
외부 API ──────────→ │ L: 즉시   │     │  orders_raw             │
                    │    적재   │     │  users_raw              │
                    └───────────┘     │  logs_raw               │
                    (변환 없이 빠름)  │          ↓               │
                                      │  T: DW 내부에서 변환    │
                                      │  (BigQuery SQL /        │
                                      │   Snowflake SQL /       │
                                      │   dbt 모델)             │
                                      │          ↓               │
                                      │  Curated Layer (정제됨) │
                                      │  orders_clean           │
                                      │  user_metrics           │
                                      │  daily_revenue          │
                                      └─────────────────────────┘

핵심: 원본 데이터가 DW 안에 항상 보존됨
      변환은 DW의 컴퓨팅 파워를 활용

[ELT 부상의 배경]

변화 1: 클라우드 DW의 컴퓨팅 파워 폭증
  BigQuery, Snowflake, Redshift는 수천 개 노드가
  병렬로 SQL을 처리 → 수TB 데이터도 분 단위 변환
  기존 ETL 서버 한 대보다 수백 배 빠름

변화 2: 스토리지 비용의 급격한 하락
  온프레미스 DW: GB당 수천~수만 원
  클라우드 오브젝트 스토리지: GB당 수십 원
  → 원본 데이터 전체를 저장해도 비용 부담 없음

변화 3: Column-store 아키텍처 최적화
  BigQuery, Snowflake는 컬럼형 저장 + 대규모 병렬 처리
  변환용 SQL이 수십억 행에서도 초~분 단위로 완료
  → DW 내부에서 변환하는 것이 오히려 더 효율적

[스키마 온 라이트 vs 스키마 온 리드]

ETL (스키마 온 라이트, Schema-on-Write):
  데이터 저장 전에 스키마(구조) 결정 필수
  → 나중에 "이 컬럼도 필요했는데" 하면 재처리 필요
  → 사전에 모든 사용 케이스를 예상해야 함

ELT (스키마 온 리드, Schema-on-Read):
  원본을 그대로 저장 후 사용 시점에 구조 결정
  → 나중에 어떤 분석이든 원본에서 새로 변환 가능
  → ML 학습, 새로운 KPI 추가 등 유연한 대응

예시:
  오늘: 일별 매출 집계 테이블 생성 (SQL 변환)
  내일: 동일 원본으로 고객 이탈 예측 피처 생성 (동일 원본 재활용)
  다음 주: 원본에서 새로운 이상 거래 탐지 모델 피처 추가
  → ELT는 원본이 보존되어 있어 언제든 가능

ETL:
소스 → [외부 변환 엔진] → DW
         ↑
  Apache Spark, Informatica, SSIS 등
  전용 서버에서 변환 후 DW에 깨끗한 데이터만 전달

ELT:
소스 → DW Raw Layer → [DW 내부 변환] → DW Curated Layer
                              ↑
                   BigQuery SQL, dbt, Snowflake SQL
                   DW 자체 엔진으로 변환

# ETL: 변환 후 원본 데이터가 중간 처리 서버에서 삭제됨
def etl_pipeline(source_records: list) -> list:
    """
    ETL 방식: DW에는 변환된 데이터만 저장
    """
    # E: 추출
    raw_data = extract(source_records)

    # T: 변환 (중간 서버에서 처리)
    transformed = []
    for row in raw_data:
        clean_row = {
            'order_id':    row['id'],
            'amount_krw':  int(row['amount_usd'] * 1350),  # 환율 변환
            'order_date':  row['created_at'][:10],          # 날짜 정규화
            'status':      row['status'].upper(),
        }
        transformed.append(clean_row)

    # L: 적재 후 중간 파일 삭제
    load_to_dw(transformed)
    delete_staging_files()  # 원본 흔적 제거

    return transformed  # DW에는 변환본만 존재


# ELT: 원본 + 변환본 모두 DW에 보존
def elt_pipeline(source_records: list) -> None:
    """
    ELT 방식: 원본 그대로 먼저 적재, 변환은 DW SQL로
    """
    # E: 추출
    raw_data = extract(source_records)

    # L: 원본 그대로 즉시 적재 (변환 없음)
    load_raw_to_dw(raw_data, table='orders_raw')
    # → orders_raw 테이블에 원본이 영구 보존

    # T: DW 안에서 SQL로 변환 (dbt 모델 실행)
    execute_dbt_model('orders_clean')  # 아래 SQL 실행
    # DW 내부 SQL:
    # CREATE OR REPLACE TABLE orders_clean AS
    # SELECT
    #   id AS order_id,
    #   CAST(amount_usd * 1350 AS INT64) AS amount_krw,
    #   DATE(created_at) AS order_date,
    #   UPPER(status) AS status
    # FROM orders_raw
    # WHERE amount_usd > 0  -- 이상값 제거

[대용량 데이터 처리 성능 비교]

시나리오: 1억 건 주문 데이터 변환

ETL 방식:
  소스 → ETL 서버(32코어, 256GB RAM) → DW
  처리 시간: 약 6~12시간
  병목: ETL 서버 단일 노드 한계
  스케일업: 서버 사양 업그레이드 (비용 선형 증가)

ELT 방식:
  소스 → DW 즉시 적재 → BigQuery/Snowflake 내부 SQL
  처리 시간: 약 5~30분
  병렬 처리: DW가 자동으로 수천 노드에 분산 처리
  스케일아웃: DW 클러스터 자동 확장 (서버리스)

  BigQuery 1억 행 집계 SQL 실행 예시:
  SELECT DATE(created_at), SUM(amount_usd)
  FROM orders_raw  -- 1억 행
  GROUP BY 1
  → 실제 실행 시간: 약 3~8초 (수백 슬롯 병렬 처리)

[보안 관점 ETL vs ELT]

ETL 보안 강점:
  ✅ 민감 데이터가 DW에 절대 도달하지 않음
  ✅ 마스킹·익명화를 파이프라인 레벨에서 강제
  ✅ GDPR·개인정보보호법 준수 용이
  ✅ 규정 상 원본 보존 불가 데이터에 적합

ELT 보안 고려사항:
  ⚠️ 원본 데이터가 클라우드 DW에 저장됨
     → DW 접근 권한 관리가 매우 중요
  ⚠️ 민감 컬럼 row-level / column-level 보안 필요
  ✅ 현대 DW(Snowflake, BigQuery)의 세밀한 접근 제어로 해결 가능
  ✅ 원본 보존으로 감사(Audit) 로그 추적 용이

# ETL 파이프라인: MySQL → Transform → BigQuery
# Apache Airflow DAG 구조

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

# DAG 기본 설정
default_args = {
    'owner':            'data-team',
    'retries':          3,
    'retry_delay':      timedelta(minutes=5),
    'email_on_failure': True,
}

def extract_from_mysql(**context) -> str:
    """
    E: MySQL에서 어제 주문 데이터 추출
    """
    mysql_hook = MySqlHook(mysql_conn_id='mysql_orders')
    execution_date = context['ds']   # 실행 날짜 (YYYY-MM-DD)

    sql = f"""
        SELECT
            id, user_id, amount_usd, currency,
            status, created_at, updated_at
        FROM orders
        WHERE DATE(created_at) = '{execution_date}'
          AND is_deleted = 0
    """
    df = mysql_hook.get_pandas_df(sql)

    # Staging 경로에 임시 저장 (XCom 또는 GCS)
    staging_path = f'/tmp/orders_{execution_date}.parquet'
    df.to_parquet(staging_path, index=False)

    print(f"추출 완료: {len(df)}건 → {staging_path}")
    return staging_path


def transform_orders(**context) -> str:
    """
    T: 비즈니스 규칙 적용 변환 (외부 서버에서 처리)
    """
    execution_date = context['ds']
    staging_path   = f'/tmp/orders_{execution_date}.parquet'

    df = pd.read_parquet(staging_path)

    # ── 변환 규칙 적용 ──────────────────────────────────────

    # 1. 환율 변환 (USD → KRW)
    EXCHANGE_RATES = {'USD': 1350, 'EUR': 1480, 'JPY': 9.2}
    df['amount_krw'] = df.apply(
        lambda r: int(r['amount_usd'] * EXCHANGE_RATES.get(r['currency'], 1350)),
        axis=1
    )

    # 2. 상태 코드 정규화
    STATUS_MAP = {
        'completed': 'COMPLETED',
        'complete':  'COMPLETED',   # 레거시 값 정규화
        'cancel':    'CANCELLED',
        'cancelled': 'CANCELLED',
        'pending':   'PENDING',
        'refund':    'REFUNDED',
    }
    df['status_normalized'] = df['status'].str.lower().map(STATUS_MAP).fillna('UNKNOWN')

    # 3. 날짜 파생 컬럼 생성
    df['order_date']  = pd.to_datetime(df['created_at']).dt.date
    df['order_hour']  = pd.to_datetime(df['created_at']).dt.hour
    df['order_month'] = pd.to_datetime(df['created_at']).dt.to_period('M').astype(str)

    # 4. 이상값 제거
    df = df[df['amount_usd'] > 0]        # 음수·0 금액 제거
    df = df.drop_duplicates(subset='id') # 중복 주문 제거

    # 5. 최종 컬럼 선택 (DW 스키마에 맞게)
    final_df = df[[
        'id', 'user_id', 'amount_usd', 'amount_krw',
        'status_normalized', 'order_date', 'order_hour', 'order_month'
    ]].rename(columns={
        'id':                 'order_id',
        'status_normalized':  'status'
    })

    # 변환된 데이터 저장
    transformed_path = f'/tmp/orders_clean_{execution_date}.parquet'
    final_df.to_parquet(transformed_path, index=False)

    print(f"변환 완료: {len(final_df)}건 → {transformed_path}")
    return transformed_path


def load_to_bigquery(**context) -> None:
    """
    L: 변환 완료된 데이터를 BigQuery에 적재
    """
    execution_date   = context['ds']
    transformed_path = f'/tmp/orders_clean_{execution_date}.parquet'

    df = pd.read_parquet(transformed_path)
    bq_hook = BigQueryHook(gcp_conn_id='google_cloud_default')

    # BigQuery에 적재 (해당 날짜 파티션 덮어쓰기)
    bq_hook.insert_rows_from_dataframe(
        dataframe    = df,
        project_id   = 'my-project',
        dataset_id   = 'analytics',
        table_id     = f'orders_daily${execution_date.replace("-", "")}',
        chunk_size   = 10000
    )
    print(f"적재 완료: BigQuery analytics.orders_daily")


# ── Airflow DAG 정의 ───────────────────────────────────────────
with DAG(
    dag_id            = 'etl_orders_daily',
    default_args      = default_args,
    schedule_interval = '0 2 * * *',   # 매일 새벽 2시 실행
    start_date        = datetime(2024, 1, 1),
    catchup           = False,
    tags              = ['etl', 'orders']
) as dag:

    t_extract   = PythonOperator(task_id='extract',   python_callable=extract_from_mysql)
    t_transform = PythonOperator(task_id='transform', python_callable=transform_orders)
    t_load      = PythonOperator(task_id='load',      python_callable=load_to_bigquery)

    # ETL 순서: 추출 → 변환 → 적재
    t_extract >> t_transform >> t_load

-- ELT 방식: 원본은 BigQuery에 그대로 적재 완료
-- dbt가 BigQuery 안에서 SQL로 변환 레이어 구성

-- ── 파일: models/staging/stg_orders.sql ─────────────────────
-- Staging 레이어: 원본에서 기본 정제만 수행
{{
  config(
    materialized = 'view',          -- 뷰로 생성 (저장 비용 없음)
    schema       = 'staging'
  )
}}

SELECT
    id                                        AS order_id,
    user_id,
    CAST(amount_usd AS FLOAT64)               AS amount_usd,
    UPPER(TRIM(currency))                     AS currency,
    LOWER(TRIM(status))                       AS status_raw,
    TIMESTAMP(created_at)                     AS created_at,
    TIMESTAMP(updated_at)                     AS updated_at,
    CURRENT_TIMESTAMP()                       AS _loaded_at   -- 메타데이터
FROM {{ source('raw', 'orders_raw') }}        -- 원본 테이블 참조
WHERE id IS NOT NULL
  AND amount_usd IS NOT NULL


-- ── 파일: models/marts/orders_clean.sql ─────────────────────
-- Mart 레이어: 비즈니스 분석용 최종 테이블
{{
  config(
    materialized  = 'incremental',  -- 증분 처리 (변경분만 갱신)
    schema        = 'marts',
    partition_by  = {'field': 'order_date', 'data_type': 'date'},
    cluster_by    = ['status', 'order_month'],
    unique_key    = 'order_id'
  )
}}

WITH exchange_rates AS (
    -- 환율 테이블 (별도 소스)
    SELECT currency, rate_to_krw
    FROM {{ ref('exchange_rates_daily') }}
    WHERE rate_date = CURRENT_DATE()
),

status_mapping AS (
    -- 상태 코드 정규화 매핑
    SELECT *
    FROM (VALUES
        ('completed', 'COMPLETED'),
        ('complete',  'COMPLETED'),
        ('cancel',    'CANCELLED'),
        ('cancelled', 'CANCELLED'),
        ('pending',   'PENDING'),
        ('refund',    'REFUNDED')
    ) AS t(status_raw, status_clean)
),

transformed AS (
    SELECT
        o.order_id,
        o.user_id,
        o.amount_usd,
        -- 환율 적용 KRW 변환
        ROUND(o.amount_usd * COALESCE(er.rate_to_krw, 1350)) AS amount_krw,
        -- 상태 정규화
        COALESCE(sm.status_clean, 'UNKNOWN')                  AS status,
        -- 날짜 파생 컬럼
        DATE(o.created_at)                                    AS order_date,
        EXTRACT(HOUR FROM o.created_at)                       AS order_hour,
        FORMAT_DATE('%Y-%m', DATE(o.created_at))              AS order_month,
        o.created_at,
        o._loaded_at
    FROM {{ ref('stg_orders') }}      o   -- staging 뷰 참조
    LEFT JOIN exchange_rates          er  ON o.currency = er.currency
    LEFT JOIN status_mapping          sm  ON o.status_raw = sm.status_raw
    WHERE o.amount_usd > 0            -- 이상값 제거
)

SELECT * FROM transformed

{% if is_incremental() %}
    -- 증분 모드: 마지막 실행 이후 변경분만 처리
    WHERE created_at > (SELECT MAX(created_at) FROM {{ this }})
{% endif %}

# dbt 소스 정의 파일: models/sources.yml
# ELT에서 원본 데이터가 어디 있는지 선언

version: 2

sources:
  - name: raw
    database: my-project
    schema: raw_data
    description: "Fivetran/Airbyte가 적재한 원본 데이터"
    tables:
      - name: orders_raw
        description: "MySQL orders 테이블 원본 (미변환)"
        columns:
          - name: id
            description: "주문 고유 ID"
            tests:
              - not_null
              - unique
          - name: amount_usd
            tests:
              - not_null

[ETL vs ELT 선택 가이드]

클라우드 DW(BigQuery, Snowflake, Redshift)를 사용하는가?
    │
    ├── YES → ELT 우선 검토
    │           └── 민감 데이터(주민번호, 카드번호)가 포함되는가?
    │                   ├── YES → 적재 전 마스킹 ETL 레이어 추가
    │                   │          (하이브리드: ETL for PII + ELT for rest)
    │                   └── NO  → 순수 ELT 적용
    │
    └── NO (온프레미스, 레거시 DW)
            ↓
        데이터 볼륨이 일 1TB를 초과하는가?
            ├── YES → Spark 기반 ETL + 클라우드 이전 검토
            └── NO  → 전통 ETL (Informatica, SSIS 등)

공통 체크리스트:
  □ 원본 데이터 재처리 가능성이 있는가?  → ELT 유리
  □ 규정 상 원본 데이터 저장 금지인가?   → ETL 필수
  □ 분석팀이 SQL을 직접 다루는가?        → ELT 유리
  □ 실시간 스트리밍 처리가 필요한가?     → Kafka + Spark ETL 검토

[2024년 표준 데이터 플랫폼 구성]

소스 시스템
  MySQL, PostgreSQL, Salesforce, Stripe, S3...
          ↓
[E + L 단계: 커넥터 도구]
  Fivetran / Airbyte (관리형 커넥터)
  → 소스별 커넥터 제공, 스케줄 관리 자동화
  → 변환 없이 Raw 그대로 DW에 적재
          ↓
[클라우드 DW – Raw Layer]
  BigQuery / Snowflake / Databricks
  raw_data 스키마: orders_raw, users_raw, events_raw
          ↓
[T 단계: 변환 레이어]
  dbt (data build tool)
  → SQL 기반 변환 모델 정의
  → 테스트·문서화·버전관리 내장
  → Staging → Intermediate → Marts 레이어 구조
          ↓
[분석·소비 레이어]
  BI 도구: Looker, Tableau, Metabase
  ML 플랫폼: Vertex AI, SageMaker
  역방향 ETL: Census, Hightouch (DW → SaaS 앱 동기화)
          ↓
[오케스트레이션]
  Apache Airflow / Prefect / Dagster
  → 전체 파이프라인 스케줄·모니터링·알림

[메달리온 아키텍처 – ELT 기반 레이어 설계]

Bronze Layer (원본)    ← E + L 단계 결과
  raw_data.orders_raw
  raw_data.users_raw
  → 원본 그대로 보존, 절대 수정 금지
  → 파티션: 적재 일자 기준
         ↓
Silver Layer (정제)    ← T 단계 1차 결과 (dbt staging)
  staging.orders_clean
  staging.users_clean
  → 타입 변환, 중복 제거, 기본 정제
  → 비즈니스 규칙 최소화
         ↓
Gold Layer (분석 최적화) ← T 단계 2차 결과 (dbt marts)
  marts.daily_revenue
  marts.user_cohort
  marts.product_performance
  → 집계, 비즈니스 KPI 반영
  → BI·대시보드 직접 연결