データエンジニアリング完全ガイド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ベースのモダン UI
   - リアルタイムログストリーミング
   - DAG可視化の改善
   - ダークモード対応

2. イベント駆動スケジューリング
   - Dataset-awareスケジューリング（データ到着時に自動トリガー）
   - 外部トリガーの強化

3. TaskFlow API 2.0
   - デコレータベースのPythonネイティブDAG記述
   - 動的タスクマッピングの改善

4. Edge Labels & DAGバージョニング
   - 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以上突破、デファクトスタンダードの地位確立

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/性能

従来のアーキテクチャ：
[ソース] → [Data Lake (S3)] → [ETL] → [Data Warehouse (Redshift)]
                                         高コスト、重複ストレージ

レイクハウスアーキテクチャ：
[ソース] → [オブジェクトストレージ (S3)] + [テーブルフォーマット (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.1秒（CHの2.2倍高速）      │
│ ClickHouse     █████████  4.6秒                  │
│ Druid          ████████████████████  18.7秒      │
│ Pinot          ██████████████████  16.3秒        │
└──────────────────────────────────────────────────┘

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レイヤー     │
                    │（統合ガバナンス/カタログ）   │
                    └──────────┬───────────────┘
            ┌──────────────────┼──────────────────┐
            │                  │                  │
     ┌──────┴──────┐   ┌──────┴──────┐   ┌──────┴──────┐
     │注文ドメイン  │   │顧客ドメイン  │   │商品ドメイン  │
     │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 Feature Storeの同期
        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）