데이터 엔지니어링 완전 가이드 2025: Flink vs Spark, dbt, Iceberg, Airflow — 현대 데이터 스택 총정리

[데이터 소스] → [수집/CDC] → [스트림 처리] → [스토리지/레이크하우스] → [변환] → [분석/서빙]
     |              |             |                  |                    |           |
  DB, API,     Airbyte,      Flink,          Iceberg/Delta,           dbt,     ClickHouse,
  IoT, SaaS   Debezium,      Spark           S3/GCS/ADLS          Spark SQL   Pinot, Druid
               Fivetran      Streaming                                         StarRocks

평가 기준 체크리스트:
1. 지연 시간 요구사항 (배치 vs 니어리얼타임 vs 실시간)
2. 데이터 볼륨 (GB/일 vs TB/일 vs PB/일)
3. 팀 규모와 역량 (SQL 중심 vs 코드 중심)
4. 벤더 종속성 허용 범위
5. 예산 (오픈소스 vs 매니지드 서비스)
6. 규제 요구사항 (데이터 거버넌스, 리니지)

[이벤트 스트림] ──→ [윈도우] ──→ [집계/변환] ──→ [싱크]
                      |
              ┌───────┼───────┐
              │       │       │
          Tumbling  Sliding  Session
           Window   Window   Window
           (고정)   (슬라이딩) (세션)

// Flink - 실시간 사기 탐지 예제
DataStream<Transaction> transactions = env
    .addSource(new FlinkKafkaConsumer<>("transactions", schema, props));

DataStream<Alert> alerts = transactions
    .keyBy(Transaction::getAccountId)
    .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
    .aggregate(new FraudScoreAggregator())
    .filter(score -> score.getValue() > THRESHOLD);

alerts.addSink(new AlertSink());

-- Flink SQL: 실시간 매출 집계
CREATE TABLE orders (
    order_id STRING,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'format' = 'json'
);

SELECT
    TUMBLE_START(order_time, INTERVAL '1' MINUTE) AS window_start,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue
FROM orders
GROUP BY TUMBLE(order_time, INTERVAL '1' MINUTE);

# Spark Structured Streaming - 실시간 주문 분석
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, sum, count

spark = SparkSession.builder.appName("OrderAnalytics").getOrCreate()

orders = spark.readStream \
    .format("kafka") \
    .option("subscribe", "orders") \
    .load()

order_stats = orders \
    .withWatermark("order_time", "5 minutes") \
    .groupBy(window("order_time", "1 minute")) \
    .agg(
        count("*").alias("order_count"),
        sum("amount").alias("total_revenue")
    )

order_stats.writeStream \
    .format("iceberg") \
    .outputMode("append") \
    .option("checkpointLocation", "/checkpoint/orders") \
    .toTable("catalog.db.order_stats")

# Apache Beam - 멀티 러너 파이프라인
import apache_beam as beam

with beam.Pipeline(options=pipeline_options) as p:
    (p
     | 'ReadKafka' >> beam.io.ReadFromKafka(
         consumer_config=kafka_config,
         topics=['orders'])
     | 'ParseJSON' >> beam.Map(parse_order)
     | 'WindowInto' >> beam.WindowInto(
         beam.window.FixedWindows(60))  # 1분 윈도우
     | 'CountPerWindow' >> beam.combiners.Count.Globally()
     | 'WriteToGCS' >> beam.io.WriteToText('gs://bucket/output'))

처리 지연 시간 (p99):
┌────────────────────────────────────────────┐
│ Flink     ████  23ms                       │
│ Spark     ████████████████  450ms           │
│ Beam/Flink████  25ms                       │
│ Beam/Spark████████████████  460ms           │
└────────────────────────────────────────────┘

초당 처리량 (이벤트/초):
┌────────────────────────────────────────────┐
│ Flink     ████████████████████  5.2M       │
│ Spark     ████████████████  3.8M           │
│ Beam/Flink██████████████████  4.8M         │
└────────────────────────────────────────────┘

Airflow 3.0 핵심 업그레이드:

1. React 기반 Modern UI
   - 실시간 로그 스트리밍
   - DAG 시각화 개선
   - 다크 모드 지원

2. Event-Driven 스케줄링
   - Dataset-aware 스케줄링 (데이터 도착 시 자동 트리거)
   - External Trigger 강화

3. TaskFlow API 2.0
   - 데코레이터 기반 Python 네이티브 DAG 작성
   - 동적 태스크 매핑 개선

4. Edge Labels & DAG Versioning
   - DAG 변경 이력 추적
   - A/B 테스트 DAG 지원

5. 보안 강화
   - RBAC 개선
   - Secret Backend 통합 확장

# Airflow 3.0 - TaskFlow API 예제
from airflow.decorators import dag, task
from datetime import datetime

@dag(schedule="@daily", start_date=datetime(2025, 1, 1), catchup=False)
def data_pipeline():

    @task()
    def extract():
        """S3에서 원본 데이터 추출"""
        import boto3
        # 데이터 추출 로직
        return {"records": 15000, "source": "s3://datalake/raw/"}

    @task()
    def transform(data: dict):
        """dbt를 통한 데이터 변환"""
        import subprocess
        subprocess.run(["dbt", "run", "--select", "staging+"])
        return {"transformed": data["records"], "models": 12}

    @task()
    def load(data: dict):
        """ClickHouse에 결과 적재"""
        # ClickHouse 적재 로직
        return {"loaded": data["transformed"]}

    @task()
    def notify(result: dict):
        """Slack 알림 전송"""
        # Slack 알림 로직
        pass

    raw = extract()
    transformed = transform(raw)
    loaded = load(transformed)
    notify(loaded)

data_pipeline()

# Dagster - Software-Defined Assets 예제
from dagster import asset, AssetExecutionContext
from dagster_dbt import DbtCliResource, dbt_assets

@asset(
    description="원본 주문 데이터를 S3에서 추출",
    group_name="raw",
    compute_kind="python"
)
def raw_orders(context: AssetExecutionContext):
    """S3에서 주문 데이터 추출"""
    import pandas as pd
    df = pd.read_parquet("s3://datalake/raw/orders/")
    context.log.info(f"Extracted {len(df)} orders")
    return df

@asset(
    description="주문 데이터 품질 검증",
    group_name="validated",
    compute_kind="great_expectations"
)
def validated_orders(raw_orders):
    """Great Expectations로 데이터 품질 검증"""
    # 품질 검증 로직
    assert raw_orders["amount"].min() >= 0, "음수 금액 발견"
    return raw_orders

# dbt 자산 자동 통합
@dbt_assets(manifest=dbt_manifest_path)
def dbt_models(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

# Prefect - 간결한 파이프라인
from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(retries=3, retry_delay_seconds=60,
      cache_key_fn=task_input_hash,
      cache_expiration=timedelta(hours=1))
def extract_data(source: str) -> dict:
    """자동 재시도와 캐싱이 적용된 추출"""
    # 데이터 추출
    return {"data": [...], "count": 15000}

@task(log_prints=True)
def transform_data(raw: dict) -> dict:
    """변환 로직"""
    print(f"Processing {raw['count']} records")
    return {"transformed": raw["count"]}

@flow(name="daily-etl", log_prints=True)
def daily_pipeline():
    raw = extract_data("s3://datalake/raw/")
    result = transform_data(raw)
    print(f"Pipeline complete: {result}")

# 실행
daily_pipeline()

dbt 마일스톤:
- 2016: fishtown analytics 설립
- 2020: dbt Cloud 출시
- 2022: dbt Labs로 리브랜딩, Series D $222M
- 2023: MetricFlow 오픈소스화
- 2024: dbt Cloud에 Semantic Layer GA
- 2025: ARR $100M+ 돌파, de facto standard 지위 확립

dbt_project/
├── dbt_project.yml
├── packages.yml
├── profiles.yml
├── models/
│   ├── staging/           # 원본 데이터 정제
│   │   ├── stg_orders.sql
│   │   ├── stg_customers.sql
│   │   └── _staging.yml   # 테스트 & 문서
│   ├── intermediate/      # 비즈니스 로직 조합
│   │   ├── int_order_items.sql
│   │   └── _intermediate.yml
│   ├── marts/             # 최종 분석용 테이블
│   │   ├── finance/
│   │   │   ├── fct_revenue.sql
│   │   │   └── dim_customers.sql
│   │   └── marketing/
│   │       └── fct_campaigns.sql
│   └── metrics/           # MetricFlow 메트릭
│       └── revenue.yml
├── tests/                 # 커스텀 테스트
│   └── assert_positive_revenue.sql
├── macros/                # 재사용 가능한 SQL 매크로
│   └── cents_to_dollars.sql
├── seeds/                 # 정적 데이터 (CSV)
│   └── country_codes.csv
└── snapshots/             # SCD Type 2 스냅샷
    └── snap_customers.sql

-- models/staging/stg_orders.sql
WITH source AS (
    SELECT * FROM {{ source('raw', 'orders') }}
),

renamed AS (
    SELECT
        id AS order_id,
        user_id AS customer_id,
        status AS order_status,
        amount_cents / 100.0 AS amount_dollars,
        created_at AS ordered_at
    FROM source
    WHERE status != 'cancelled'
)

SELECT * FROM renamed

-- models/marts/finance/fct_revenue.sql
WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

customers AS (
    SELECT * FROM {{ ref('dim_customers') }}
),

final AS (
    SELECT
        o.order_id,
        o.customer_id,
        c.customer_segment,
        o.amount_dollars,
        o.ordered_at,
        DATE_TRUNC('month', o.ordered_at) AS order_month
    FROM orders o
    JOIN customers c ON o.customer_id = c.customer_id
    WHERE o.order_status = 'completed'
)

SELECT * FROM final

# models/staging/_staging.yml
version: 2

models:
  - name: stg_orders
    description: '정제된 주문 데이터'
    columns:
      - name: order_id
        description: '주문 고유 식별자'
        tests:
          - unique
          - not_null
      - name: amount_dollars
        description: '주문 금액 (달러)'
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 100000

# models/metrics/revenue.yml
semantic_models:
  - name: orders
    defaults:
      agg_time_dimension: ordered_at
    model: ref('fct_revenue')
    entities:
      - name: order_id
        type: primary
      - name: customer_id
        type: foreign
    measures:
      - name: revenue
        agg: sum
        expr: amount_dollars
      - name: order_count
        agg: count
        expr: order_id
    dimensions:
      - name: ordered_at
        type: time
      - name: customer_segment
        type: categorical

metrics:
  - name: monthly_revenue
    type: simple
    type_params:
      measure: revenue
  - name: revenue_per_customer
    type: derived
    type_params:
      expr: monthly_revenue / active_customers
      metrics:
        - name: monthly_revenue
        - name: active_customers

dbt 프로젝트 베스트 프랙티스 10가지:

1. 계층 구조 엄수: staging → intermediate → marts
2. staging에서 모든 이름 변경 및 타입 캐스팅 완료
3. 모든 모델에 unique + not_null 테스트 적용
4. source freshness 체크로 데이터 신선도 모니터링
5. 증분 모델(incremental)로 대용량 테이블 최적화
6. packages.yml로 dbt-utils, dbt-expectations 활용
7. pre-commit 훅으로 SQL 린팅 (sqlfluff)
8. CI에서 slim CI (state:modified+) 활용
9. 환경별 프로파일 분리 (dev / staging / prod)
10. Semantic Layer로 메트릭 중앙 관리

레이크하우스 = 데이터 레이크의 저비용 스토리지 + 웨어하우스의 ACID/성능

전통 아키텍처:
[Sources] → [Data Lake (S3)] → [ETL] → [Data Warehouse (Redshift)]
                                         비용 높음, 중복 저장

레이크하우스 아키텍처:
[Sources] → [Object Storage (S3)] + [Table Format (Iceberg)]
             저비용                   ACID, 스키마 진화, 타임 트래블
             └──→ [분석 엔진 (Spark/Trino/Flink)] 직접 쿼리

-- Iceberg 테이블 생성 및 활용
CREATE TABLE catalog.db.orders (
    order_id    BIGINT,
    customer_id BIGINT,
    amount      DECIMAL(10,2),
    status      STRING,
    ordered_at  TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(ordered_at));

-- 타임 트래블: 어제 시점의 데이터 조회
SELECT * FROM catalog.db.orders
VERSION AS OF '2025-03-21T00:00:00';

-- 스냅샷 기반 증분 읽기
SELECT * FROM catalog.db.orders
WHERE ordered_at > (
    SELECT max(ordered_at)
    FROM catalog.db.orders
    VERSION AS OF 'snapshot_id_123'
);

# Delta Lake - UniForm으로 Iceberg 호환
from delta.tables import DeltaTable

# Delta 테이블 생성 (UniForm 활성화)
spark.sql("""
    CREATE TABLE catalog.db.events (
        event_id BIGINT,
        event_type STRING,
        payload STRING,
        event_time TIMESTAMP
    )
    USING DELTA
    TBLPROPERTIES (
        'delta.universalFormat.enabledFormats' = 'iceberg'
    )
""")

# Delta 테이블에 MERGE (Upsert)
delta_table = DeltaTable.forName(spark, "catalog.db.events")

delta_table.alias("target").merge(
    new_events.alias("source"),
    "target.event_id = source.event_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

-- ClickHouse - 실시간 대시보드 쿼리
CREATE TABLE events (
    event_id UInt64,
    user_id UInt32,
    event_type LowCardinality(String),
    properties Map(String, String),
    event_time DateTime64(3)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, user_id, event_time);

-- Materialized View로 실시간 집계
CREATE MATERIALIZED VIEW hourly_events_mv
ENGINE = SummingMergeTree()
ORDER BY (event_type, hour)
AS SELECT
    event_type,
    toStartOfHour(event_time) AS hour,
    count() AS event_count,
    uniqHLL12(user_id) AS unique_users
FROM events
GROUP BY event_type, hour;

-- 수십억 행에서 밀리초 응답
SELECT
    event_type,
    sum(event_count) AS total_events,
    sum(unique_users) AS total_users
FROM hourly_events_mv
WHERE hour >= now() - INTERVAL 24 HOUR
GROUP BY event_type
ORDER BY total_events DESC
LIMIT 20;

TPC-H 10GB 쿼리 성능 (초, 낮을수록 좋음):
┌──────────────────────────────────────────────────┐
│ StarRocks 3.x  ████  2.1s (2.2x faster than CH) │
│ ClickHouse     █████████  4.6s                   │
│ Druid          ████████████████████  18.7s       │
│ Pinot          ██████████████████  16.3s         │
└──────────────────────────────────────────────────┘

1. 도메인 소유권 (Domain Ownership)
   → 각 도메인 팀이 자신의 데이터를 소유

2. 제품으로서의 데이터 (Data as a Product)
   → 데이터에 SLA, 문서, 품질 보증 적용

3. 셀프 서비스 플랫폼 (Self-Serve Platform)
   → 인프라 팀이 데이터 플랫폼 도구 제공

4. 연합 거버넌스 (Federated Governance)
   → 전사 표준 + 도메인 자율성의 균형

# data-contract.yml 예시
dataContractSpecification: 0.9.3
id: orders-contract
info:
  title: '주문 데이터 계약'
  version: 2.1.0
  owner: order-domain-team
  contact:
    name: '주문팀'
    email: orders@company.com

servers:
  production:
    type: iceberg
    catalog: polaris
    database: orders_db
    table: orders

models:
  orders:
    description: '완료된 주문 데이터'
    fields:
      order_id:
        type: bigint
        required: true
        unique: true
        description: '주문 고유 식별자'
      customer_id:
        type: bigint
        required: true
      amount:
        type: decimal
        required: true
        minimum: 0
        description: '주문 금액 (USD)'
      ordered_at:
        type: timestamp
        required: true

quality:
  type: SodaCL
  specification:
    checks for orders:
      - row_count > 0
      - missing_count(order_id) = 0
      - duplicate_count(order_id) = 0
      - avg(amount) between 10 and 500

sla:
  freshness: 1 hour
  availability: 99.9%
  latency: p99 < 500ms

Data Mesh + Fabric 하이브리드:

                    ┌─────────────────────┐
                    │   Data Fabric Layer  │
                    │  (통합 거버넌스/카탈로그)│
                    └────────┬────────────┘
            ┌────────────────┼────────────────┐
            │                │                │
     ┌──────┴──────┐  ┌─────┴──────┐  ┌──────┴──────┐
     │ 주문 도메인  │  │ 고객 도메인  │  │ 상품 도메인  │
     │ Data Product │  │ Data Product│  │ Data Product │
     │  (Iceberg)   │  │  (Iceberg)  │  │  (Iceberg)   │
     └─────────────┘  └────────────┘  └─────────────┘

요구사항:
- 일일 활성 사용자: 500만명
- 초당 이벤트: 50,000건
- 추천 지연 시간: p99 < 100ms
- 데이터 소스: 클릭 스트림, 구매, 검색, 재고
- 피처 스토어: 실시간 + 배치 피처 결합
- A/B 테스트: 모델 버전별 성과 비교

실시간 추천 시스템 아키텍처:

[클릭/구매/검색 이벤트]
        │
        ▼
   ┌─────────┐     ┌──────────────┐
   │  Kafka  │────→│  Flink (CEP) │──→ 실시간 피처 계산
   │ Cluster │     │ 윈도우 집계   │    (최근 5분 클릭 카테고리)
   └────┬────┘     └──────┬───────┘
        │                 │
        ▼                 ▼
   ┌─────────┐     ┌──────────────┐
   │ Iceberg │     │    Redis     │──→ 실시간 피처 서빙
   │ (원본)  │     │ Feature Store│    (p99 < 5ms)
   └────┬────┘     └──────────────┘
        │                 ↑
        ▼                 │
   ┌─────────┐     ┌──────────────┐
   │   dbt   │────→│  ClickHouse  │──→ 배치 피처 계산
   │ (변환)  │     │  (집계/분석)  │    (30일 구매 패턴)
   └────┬────┘     └──────────────┘
        │
        ▼
   ┌─────────────────┐     ┌──────────────┐
   │ ML Model Serving │←───│ Feature Store │
   │ (TorchServe/     │    │ (Redis+DuckDB)│
   │  Triton)         │    └──────────────┘
   └────────┬────────┘
            │
            ▼
   ┌─────────────────┐
   │   API Gateway   │──→ 사용자에게 추천 결과 제공
   │  (p99 < 100ms)  │
   └─────────────────┘

// 실시간 사용자 행동 피처 계산
DataStream<UserFeature> realtimeFeatures = clickEvents
    .keyBy(ClickEvent::getUserId)
    .window(SlidingEventTimeWindows.of(
        Time.minutes(5), Time.minutes(1)))
    .aggregate(new UserBehaviorAggregator());

// Redis 피처 스토어에 저장
realtimeFeatures.addSink(
    new RedisSink<>(redisConfig, new FeatureRedisMapper()));

-- models/features/user_purchase_patterns.sql
-- 30일 구매 패턴 피처 (배치)
WITH purchase_stats AS (
    SELECT
        customer_id,
        COUNT(*) AS purchase_count_30d,
        SUM(amount_dollars) AS total_spend_30d,
        AVG(amount_dollars) AS avg_order_value_30d,
        COUNT(DISTINCT category) AS unique_categories_30d,
        MAX(ordered_at) AS last_purchase_at
    FROM {{ ref('fct_revenue') }}
    WHERE ordered_at >= CURRENT_DATE - INTERVAL '30' DAY
    GROUP BY customer_id
)

SELECT
    customer_id,
    purchase_count_30d,
    total_spend_30d,
    avg_order_value_30d,
    unique_categories_30d,
    DATEDIFF('day', last_purchase_at, CURRENT_DATE) AS days_since_last_purchase
FROM purchase_stats

# 추천 시스템 일일 파이프라인
@dag(schedule="0 2 * * *", catchup=False)
def recommendation_pipeline():

    @task()
    def run_dbt_features():
        """dbt로 배치 피처 생성"""
        import subprocess
        result = subprocess.run(
            ["dbt", "run", "--select", "features+"],
            capture_output=True, text=True
        )
        return result.returncode == 0

    @task()
    def export_to_feature_store(dbt_success: bool):
        """배치 피처를 Feature Store로 내보내기"""
        if not dbt_success:
            raise Exception("dbt 실행 실패")
        # ClickHouse → Redis 피처 스토어 동기화
        pass

    @task()
    def retrain_model():
        """ML 모델 재훈련"""
        # 일일 모델 재훈련 로직
        pass

    @task()
    def validate_model(retrain_result):
        """모델 성능 검증"""
        # A/B 테스트 메트릭 비교
        pass

    dbt_result = run_dbt_features()
    export = export_to_feature_store(dbt_result)
    retrain = retrain_model()
    validate_model(retrain)

recommendation_pipeline()

월간 예상 비용 (AWS 기준, 일일 500만 유저):

| 구성 요소      | 인스턴스 타입        | 월 비용      |
|---------------|--------------------|-----------  |
| Kafka (MSK)   | kafka.m5.2xlarge x3| 약 2,100달러 |
| Flink (EMR)   | m5.2xlarge x4      | 약 2,800달러 |
| Iceberg (S3)  | 50TB 스토리지       | 약 1,150달러 |
| ClickHouse    | m5.4xlarge x3      | 약 4,200달러 |
| Redis         | r6g.2xlarge x2     | 약 1,800달러 |
| Airflow (MWAA)| mw1.medium         | 약 350달러   |

총 예상 비용: 약 12,400달러/월

비용 최적화 전략:
1. Spot 인스턴스 활용 (Flink 워커) → 60% 절감
2. Iceberg 컴팩션 최적화 → 스토리지 30% 절감
3. ClickHouse TTL 정책 → 오래된 데이터 자동 삭제
4. Reserved Instance → 장기 워크로드 40% 절감

12개월 데이터 엔지니어 학습 경로:

[1-3개월: 기초]
├── SQL 고급 (윈도우 함수, CTE, 최적화)
├── Python 데이터 처리 (pandas, polars)
├── Linux / Docker 기초
└── Git / CI/CD 기초

[4-6개월: 핵심 도구]
├── Apache Spark (DataFrame API, SparkSQL)
├── dbt (프로젝트 구조, 테스트, 문서화)
├── Airflow (DAG 작성, 오퍼레이터)
└── Kafka 기초 (프로듀서, 컨슈머, 토픽)

[7-9개월: 심화]
├── Apache Flink (스트림 처리, 상태 관리)
├── Iceberg / Delta Lake (레이크하우스)
├── ClickHouse (OLAP 분석)
└── Great Expectations (데이터 품질)

[10-12개월: 실전 & 자격증]
├── 포트폴리오 프로젝트 3개 완성
├── 자격증 1-2개 취득
├── Data Mesh / 거버넌스 학습
└── 기술 블로그 작성 시작

기술 스택: Kafka + Flink + Iceberg + dbt + ClickHouse

아키텍처:
NYC 택시 데이터(API) → Kafka → Flink(실시간 피처) → Iceberg
                                                       ↓
                                              dbt(배치 변환)
                                                       ↓
                                              ClickHouse(대시보드)

구현 포인트:
1. Flink로 실시간 지역별 수요 예측 피처 생성
2. Iceberg 파티션 진화 활용 (일별 → 시간별)
3. dbt 증분 모델로 일일 집계
4. ClickHouse Materialized View로 실시간 대시보드
5. Airflow DAG로 전체 파이프라인 오케스트레이션

기술 스택: Debezium + Kafka + Spark + Delta Lake + dbt

아키텍처:
PostgreSQL(OLTP) → Debezium(CDC) → Kafka → Spark Streaming → Delta Lake
                                                                 ↓
                                                        dbt(계층적 변환)
                                                                 ↓
                                                  Superset(BI 대시보드)

구현 포인트:
1. Debezium으로 PostgreSQL 변경 사항 실시간 캡처
2. Spark Structured Streaming으로 CDC 이벤트 처리
3. Delta Lake MERGE로 SCD Type 2 구현
4. dbt로 staging/intermediate/marts 구조화
5. Superset으로 매출/재고 실시간 대시보드

기술 스택: OpenMetadata + Soda + dbt + Iceberg + Dagster

아키텍처:
도메인 A(주문) ──→ Data Contract(YAML) ──→ Iceberg
도메인 B(고객) ──→ Data Contract(YAML) ──→ Iceberg
도메인 C(상품) ──→ Data Contract(YAML) ──→ Iceberg
                                              ↓
                              OpenMetadata(카탈로그/리니지)
                              Soda(품질 검증)
                              Dagster(오케스트레이션)

구현 포인트:
1. 3개 도메인의 Data Contract YAML 정의
2. Soda로 자동 데이터 품질 검증
3. OpenMetadata로 리니지와 카탈로그 관리
4. Dagster Software-Defined Assets로 파이프라인 구축
5. 전사 메트릭 레이어 (MetricFlow)