Split View: 데이터 엔지니어링 입문 — ETL, 데이터 웨어하우스, 스트리밍, 데이터 레이크

Raw Data Store (Data Lake)
├── Structured data (CSV, Parquet, ORC)
├── Semi-structured data (JSON, XML, Avro)
└── Unstructured data (images, logs, text)

Analytics Store (Data Warehouse)
├── Fact tables (sales, orders, clicks)
├── Dimension tables (users, products, dates)
└── Aggregate tables (daily/monthly summaries)

Bronze (Raw Data)
  → Ingested as-is from source systems
  → Minimal transformation applied

Silver (Cleansed Data)
  → Deduplication, type casting, validation complete
  → Ready for business logic but not yet aggregated

Gold (Business Data)
  → Aggregation, joining, business rules applied
  → Consumed directly by dashboards and ML models

Source → [Extract] → [Transform] → [Load] → Warehouse

Source → [Extract] → [Load] → Warehouse → [Transform]

-- dbt model example: daily revenue aggregation
-- 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 example: daily revenue aggregation
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum, count

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

# Read data
orders = spark.read.parquet("s3://data-lake/orders/")
payments = spark.read.parquet("s3://data-lake/payments/")

# Transform
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")
)

# Save results
daily_revenue.write \
    .mode("overwrite") \
    .parquet("s3://data-warehouse/daily_revenue/")

# Register DataFrames and use 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 example (Python)
from kafka import KafkaProducer
import json

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

# Publish an order event
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 example (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"Processing order: {order['order_id']}, amount: {order['amount']}")

Operational DB → [CDC] → Kafka → [Stream Processing] → Warehouse
                                                      → Search engine
                                                      → Cache

{
  "before": null,
  "after": {
    "id": 1001,
    "name": "John Doe",
    "email": "john@example.com"
  },
  "source": {
    "connector": "postgresql",
    "db": "users_db",
    "table": "users"
  },
  "op": "c",
  "ts_ms": 1681364400000
}

// Flink stream processing example (Java)
DataStream<OrderEvent> orders = env
    .addSource(new FlinkKafkaConsumer<>(
        "order-events",
        new OrderEventSchema(),
        kafkaProps
    ));

// 5-minute tumbling window aggregation
DataStream<WindowedRevenue> revenue = orders
    .keyBy(OrderEvent::getCategory)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new RevenueAggregator());

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

# Airflow DAG example
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='Daily revenue data processing pipeline',
    schedule_interval='0 2 * * *',  # Every day at 2 AM
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['revenue', 'daily'],
) as dag:

    # 1. Verify source data exists
    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 data
    extract = PythonOperator(
        task_id='extract_orders',
        python_callable=extract_orders_from_source,
    )

    # 3. Transform data
    transform = PythonOperator(
        task_id='transform_orders',
        python_callable=transform_and_aggregate,
    )

    # 4. Load to warehouse
    load = PythonOperator(
        task_id='load_to_warehouse',
        python_callable=load_to_snowflake,
    )

    # 5. Data quality validation
    validate = PythonOperator(
        task_id='validate_data_quality',
        python_callable=run_quality_checks,
    )

    # Define dependencies
    check_source >> extract >> transform >> load >> validate

-- Fact table
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)
);

-- Dimension tables
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

# Create data context
context = gx.get_context()

# Connect data source
datasource = context.sources.add_pandas("my_datasource")
data_asset = datasource.add_dataframe_asset(name="orders")

# Define expectations
batch = data_asset.build_batch_request(dataframe=df)
validator = context.get_validator(batch_request=batch)

# Data quality rules
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-.]+$"
)

# Run validation
results = validator.validate()
print(f"Success: {results.success}")

-- Snowflake RBAC example
CREATE ROLE data_analyst;
CREATE ROLE data_engineer;
CREATE ROLE data_admin;

-- Analyst role: read-only access
GRANT USAGE ON DATABASE analytics_db TO ROLE data_analyst;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics_db.gold TO ROLE data_analyst;

-- Engineer role: read/write access
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;

-- Assign roles to users
GRANT ROLE data_analyst TO USER analyst_kim;
GRANT ROLE data_engineer TO USER engineer_park;

[Source Systems]
  ├── Operational DB (PostgreSQL)  ── Debezium CDC ──┐
  ├── Web Events (Clickstream) ─── SDK ──────────────┤
  ├── External APIs ─── Airbyte ─────────────────────┤
  └── Files (CSV/Excel) ─── Airflow ─────────────────┤
                                                      │
[Message Broker]                                      │
  └── Apache Kafka <──────────────────────────────────┘
       ├── Real-time path: Flink → Real-time dashboards
       └── Batch path:     Spark → S3 (Data Lake)
                                    │
[Transformation Layer]              │
  └── dbt (Silver/Gold) <───────────┘
       │
[Warehouse]
  └── Snowflake
       ├── Gold layer → Looker/Tableau dashboards
       └── ML Feature Store → Model training

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:

    # Stage 1: Process raw data with 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',
        },
    )

    # Stage 2: Transform with dbt
    dbt_transform = DbtCloudRunJobOperator(
        task_id='dbt_transform',
        job_id=12345,
        check_interval=30,
        timeout=3600,
    )

    # Stage 3: Data quality validation
    quality_check = PythonOperator(
        task_id='data_quality_check',
        python_callable=run_great_expectations_suite,
    )

    # Stage 4: Send completion notification
    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()

# Define Kafka schema
order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id", StringType()),
    StructField("amount", DoubleType()),
    StructField("category", StringType()),
    StructField("timestamp", TimestampType()),
])

# Read streaming data from Kafka
raw_stream = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "order-events")
    .option("startingOffsets", "latest")
    .load()
)

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

# 5-minute windowed aggregation
windowed_revenue = (
    orders
    .withWatermark("timestamp", "10 minutes")
    .groupBy(
        window("timestamp", "5 minutes"),
        "category"
    )
    .agg(spark_sum("amount").alias("revenue"))
)

# Output to console (in production, write to a DB or dashboard)
query = (
    windowed_revenue.writeStream
    .outputMode("update")
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="1 minute")
    .start()
)

query.awaitTermination()