✍️ 필사 모드: 데이터 엔지니어링 입문 — ETL, 데이터 웨어하우스, 스트리밍, 데이터 레이크

원시 데이터 저장소 (Data Lake)
├── 구조화 데이터 (CSV, Parquet, ORC)
├── 반구조화 데이터 (JSON, XML, Avro)
└── 비구조화 데이터 (이미지, 로그, 텍스트)

분석용 저장소 (Data Warehouse)
├── 팩트 테이블 (매출, 주문, 클릭)
├── 디멘전 테이블 (사용자, 상품, 시간)
└── 집계 테이블 (일별/월별 요약)

Bronze (원시 데이터)
  → 소스에서 그대로 수집한 데이터
  → 최소한의 변환만 적용

Silver (정제 데이터)
  → 중복 제거, 타입 변환, 검증 완료
  → 비즈니스 로직 적용 전 상태

Gold (비즈니스 데이터)
  → 집계, 결합, 비즈니스 규칙 적용
  → 대시보드와 ML 모델에서 직접 사용

소스 → [추출] → [변환] → [적재] → 웨어하우스

소스 → [추출] → [적재] → 웨어하우스 → [변환]

-- dbt 모델 예시: 일별 매출 집계
-- models/marts/daily_revenue.sql

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

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

SELECT
    o.order_date,
    COUNT(DISTINCT o.order_id) AS total_orders,
    SUM(p.amount) AS total_revenue,
    AVG(p.amount) AS avg_order_value
FROM orders o
JOIN payments p ON o.order_id = p.order_id
WHERE p.status = 'completed'
GROUP BY o.order_date

# PySpark 예시: 일별 매출 집계
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum, count

spark = SparkSession.builder \
    .appName("DailyRevenue") \
    .getOrCreate()

# 데이터 읽기
orders = spark.read.parquet("s3://data-lake/orders/")
payments = spark.read.parquet("s3://data-lake/payments/")

# 변환
daily_revenue = (
    orders
    .join(payments, "order_id")
    .filter(col("status") == "completed")
    .groupBy("order_date")
    .agg(
        count("order_id").alias("total_orders"),
        spark_sum("amount").alias("total_revenue")
    )
    .orderBy("order_date")
)

# 결과 저장
daily_revenue.write \
    .mode("overwrite") \
    .parquet("s3://data-warehouse/daily_revenue/")

# DataFrame 등록 후 SQL 사용
orders.createOrReplaceTempView("orders")
payments.createOrReplaceTempView("payments")

result = spark.sql("""
    SELECT
        o.order_date,
        COUNT(DISTINCT o.order_id) AS total_orders,
        SUM(p.amount) AS total_revenue
    FROM orders o
    JOIN payments p ON o.order_id = p.order_id
    WHERE p.status = 'completed'
    GROUP BY o.order_date
    ORDER BY o.order_date
""")

# Kafka Producer 예시 (Python)
from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# 주문 이벤트 발행
event = {
    "order_id": "ORD-12345",
    "user_id": "USR-678",
    "amount": 45000,
    "timestamp": "2026-04-13T10:30:00Z"
}

producer.send('order-events', value=event)
producer.flush()

# Kafka Consumer 예시 (Python)
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'order-events',
    bootstrap_servers=['localhost:9092'],
    group_id='order-processing-group',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest'
)

for message in consumer:
    order = message.value
    print(f"주문 처리: {order['order_id']}, 금액: {order['amount']}")

운영 DB → [CDC] → Kafka → [Stream Processing] → 웨어하우스
                                                → 검색 엔진
                                                → 캐시

{
  "before": null,
  "after": {
    "id": 1001,
    "name": "김철수",
    "email": "kim@example.com"
  },
  "source": {
    "connector": "postgresql",
    "db": "users_db",
    "table": "users"
  },
  "op": "c",
  "ts_ms": 1681364400000
}

// Flink 스트림 처리 예시 (Java)
DataStream<OrderEvent> orders = env
    .addSource(new FlinkKafkaConsumer<>(
        "order-events",
        new OrderEventSchema(),
        kafkaProps
    ));

// 5분 단위 윈도우 집계
DataStream<WindowedRevenue> revenue = orders
    .keyBy(OrderEvent::getCategory)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new RevenueAggregator());

revenue.addSink(new JdbcSink<>(...));

# Airflow DAG 예시
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.sensors.s3_key_sensor import S3KeySensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['team@example.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='daily_revenue_pipeline',
    default_args=default_args,
    description='일별 매출 데이터 처리 파이프라인',
    schedule_interval='0 2 * * *',  # 매일 오전 2시
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['revenue', 'daily'],
) as dag:

    # 1. 소스 데이터 존재 확인
    check_source = S3KeySensor(
        task_id='check_source_data',
        bucket_name='raw-data-bucket',
        bucket_key='orders/{{ ds }}/*.parquet',
        timeout=3600,
        poke_interval=300,
    )

    # 2. 데이터 추출
    extract = PythonOperator(
        task_id='extract_orders',
        python_callable=extract_orders_from_source,
    )

    # 3. 데이터 변환
    transform = PythonOperator(
        task_id='transform_orders',
        python_callable=transform_and_aggregate,
    )

    # 4. 웨어하우스 적재
    load = PythonOperator(
        task_id='load_to_warehouse',
        python_callable=load_to_snowflake,
    )

    # 5. 데이터 품질 검증
    validate = PythonOperator(
        task_id='validate_data_quality',
        python_callable=run_quality_checks,
    )

    # 의존성 정의
    check_source >> extract >> transform >> load >> validate

-- 팩트 테이블
CREATE TABLE fact_sales (
    sale_id         BIGINT PRIMARY KEY,
    date_key        INT REFERENCES dim_date(date_key),
    product_key     INT REFERENCES dim_product(product_key),
    customer_key    INT REFERENCES dim_customer(customer_key),
    store_key       INT REFERENCES dim_store(store_key),
    quantity        INT,
    unit_price      DECIMAL(10,2),
    total_amount    DECIMAL(12,2),
    discount_amount DECIMAL(10,2)
);

-- 디멘전 테이블
CREATE TABLE dim_product (
    product_key     INT PRIMARY KEY,
    product_id      VARCHAR(50),
    product_name    VARCHAR(200),
    category        VARCHAR(100),
    subcategory     VARCHAR(100),
    brand           VARCHAR(100),
    unit_cost       DECIMAL(10,2)
);

CREATE TABLE dim_date (
    date_key        INT PRIMARY KEY,
    full_date       DATE,
    year            INT,
    quarter         INT,
    month           INT,
    week            INT,
    day_of_week     VARCHAR(20),
    is_holiday      BOOLEAN
);

import great_expectations as gx

# 데이터 컨텍스트 생성
context = gx.get_context()

# 데이터소스 연결
datasource = context.sources.add_pandas("my_datasource")
data_asset = datasource.add_dataframe_asset(name="orders")

# 기대치(Expectation) 정의
batch = data_asset.build_batch_request(dataframe=df)
validator = context.get_validator(batch_request=batch)

# 데이터 품질 규칙 정의
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_be_unique("order_id")
validator.expect_column_values_to_be_between(
    "amount", min_value=0, max_value=1000000
)
validator.expect_column_values_to_be_in_set(
    "status", ["pending", "completed", "cancelled", "refunded"]
)
validator.expect_column_values_to_match_regex(
    "email", r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
)

# 검증 실행
results = validator.validate()
print(f"성공 여부: {results.success}")

-- Snowflake RBAC 예시
CREATE ROLE data_analyst;
CREATE ROLE data_engineer;
CREATE ROLE data_admin;

-- 분석가 역할: 읽기만 가능
GRANT USAGE ON DATABASE analytics_db TO ROLE data_analyst;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics_db.gold TO ROLE data_analyst;

-- 엔지니어 역할: 읽기/쓰기 가능
GRANT ALL PRIVILEGES ON DATABASE analytics_db TO ROLE data_engineer;
GRANT ALL PRIVILEGES ON ALL SCHEMAS IN DATABASE analytics_db TO ROLE data_engineer;

-- 사용자에게 역할 할당
GRANT ROLE data_analyst TO USER analyst_kim;
GRANT ROLE data_engineer TO USER engineer_park;

[소스 시스템]
  ├── 운영 DB (PostgreSQL)  ─── Debezium CDC ──┐
  ├── 웹 이벤트 (Clickstream) ─── SDK ──────────┤
  ├── 외부 API ─── Airbyte ─────────────────────┤
  └── 파일 (CSV/Excel) ─── Airflow ─────────────┤
                                                 │
[메시지 브로커]                                   │
  └── Apache Kafka ◄────────────────────────────┘
       ├── 실시간 경로: Flink → 실시간 대시보드
       └── 배치 경로:  Spark → S3 (Data Lake)
                                │
[변환 레이어]                    │
  └── dbt (Silver/Gold) ◄──────┘
       │
[웨어하우스]
  └── Snowflake
       ├── Gold 레이어 → Looker/Tableau 대시보드
       └── ML Feature Store → 모델 학습

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=10),
    'execution_timeout': timedelta(hours=2),
}

with DAG(
    dag_id='e2e_data_pipeline',
    default_args=default_args,
    schedule_interval='0 3 * * *',
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    # 1단계: Spark로 원시 데이터 처리
    spark_process = SparkSubmitOperator(
        task_id='spark_raw_processing',
        application='s3://scripts/process_raw_data.py',
        conn_id='spark_default',
        conf={
            'spark.executor.memory': '4g',
            'spark.executor.cores': '2',
        },
    )

    # 2단계: dbt로 데이터 변환
    dbt_transform = DbtCloudRunJobOperator(
        task_id='dbt_transform',
        job_id=12345,
        check_interval=30,
        timeout=3600,
    )

    # 3단계: 데이터 품질 검증
    quality_check = PythonOperator(
        task_id='data_quality_check',
        python_callable=run_great_expectations_suite,
    )

    # 4단계: 완료 알림
    notify = SlackWebhookOperator(
        task_id='slack_notification',
        slack_webhook_conn_id='slack_webhook',
        message='Daily pipeline completed successfully.',
    )

    spark_process >> dbt_transform >> quality_check >> notify

# Spark Structured Streaming + Kafka
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = SparkSession.builder \
    .appName("RealTimeRevenue") \
    .getOrCreate()

# Kafka 스키마 정의
order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id", StringType()),
    StructField("amount", DoubleType()),
    StructField("category", StringType()),
    StructField("timestamp", TimestampType()),
])

# Kafka에서 스트리밍 데이터 읽기
raw_stream = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "order-events")
    .option("startingOffsets", "latest")
    .load()
)

# JSON 파싱
orders = (
    raw_stream
    .select(from_json(
        col("value").cast("string"),
        order_schema
    ).alias("data"))
    .select("data.*")
)

# 5분 윈도우 집계
windowed_revenue = (
    orders
    .withWatermark("timestamp", "10 minutes")
    .groupBy(
        window("timestamp", "5 minutes"),
        "category"
    )
    .agg(spark_sum("amount").alias("revenue"))
)

# 결과를 콘솔에 출력 (프로덕션에서는 DB나 대시보드로)
query = (
    windowed_revenue.writeStream
    .outputMode("update")
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="1 minute")
    .start()
)

query.awaitTermination()