데이터 엔지니어링 입문 가이드 — ETL/ELT, Apache Airflow, dbt, 웨어하우스 비교

Airflow DAG 예시 — 일일 매출 집계 파이프라인
# dags/daily_sales_pipeline.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
    'email': ['data-team@company.com'],
}

with DAG(
    'daily_sales_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # 매일 새벽 2시
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['sales', 'daily'],
) as dag:

    extract_orders = PostgresOperator(
        task_id='extract_orders',
        postgres_conn_id='source_db',
        sql='sql/extract_daily_orders.sql',
    )

    transform_sales = PythonOperator(
        task_id='transform_sales',
        python_callable=transform_sales_data,
    )

    load_warehouse = PostgresOperator(
        task_id='load_warehouse',
        postgres_conn_id='warehouse_db',
        sql='sql/load_sales_fact.sql',
    )

    run_dbt_models = BashOperator(
        task_id='run_dbt',
        bash_command='cd /opt/dbt && dbt run --select sales_mart',
    )

    # 실행 순서 정의
    extract_orders >> transform_sales >> load_warehouse >> run_dbt_models

dbt 모델 예시 — 일일 매출 마트 테이블
-- models/marts/sales/daily_sales.sql
{{ config(
    materialized='incremental',
    unique_key='sale_date',
    cluster_by=['sale_date']
) }}

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
    {% if is_incremental() %}
    WHERE order_date >= (SELECT MAX(sale_date) FROM {{ this }})
    {% endif %}
),

payments AS (
    SELECT * FROM {{ ref('stg_payments') }}
),

joined AS (
    SELECT
        o.order_date AS sale_date,
        COUNT(DISTINCT o.order_id) AS total_orders,
        COUNT(DISTINCT o.customer_id) AS unique_customers,
        SUM(p.amount) AS total_revenue,
        SUM(p.amount) / NULLIF(COUNT(DISTINCT o.order_id), 0) AS avg_order_value
    FROM orders o
    LEFT JOIN payments p ON o.order_id = p.order_id
    WHERE p.status = 'completed'
    GROUP BY 1
)

SELECT * FROM joined

데이터 엔지니어 전환 12주 로드맵
# Phase 1 (Week 1-4): SQL 고급 + 데이터 모델링
- Window Functions (ROW_NUMBER, LAG, LEAD, SUM OVER)
- CTE(Common Table Expression) 활용 복잡 쿼리
- Star Schema / Snowflake Schema 설계
- BigQuery 또는 DuckDB로 실습

# Phase 2 (Week 5-8): 파이프라인 도구
- Airflow 설치 → DAG 작성 → 스케줄링
- dbt 프로젝트 구성 → 모델 → 테스트
- Python으로 API 데이터 수집 스크립트 작성
- Docker로 Airflow + dbt 로컬 환경 구성

# Phase 3 (Week 9-12): 실전 프로젝트
- 공개 데이터셋으로 End-to-End 파이프라인 구축
  (수집 → 적재 → 변환 → 대시보드)
- Metabase 또는 Superset으로 시각화
- GitHub에 프로젝트 공개 (README, 아키텍처 다이어그램)
- 데이터 품질 테스트(dbt test) + CI/CD 통합