演習：ストリーミングパイプラインを設計しよう

from dataclasses import dataclass, asdict
from datetime import datetime
import json

@dataclass
class OrderEvent:
    order_id: str
    customer_id: int
    items: list
    total_amount: float
    status: str  # created, paid, shipped, delivered, cancelled
    region: str
    timestamp: str

@dataclass
class InventoryEvent:
    product_id: str
    warehouse_id: str
    quantity_change: int
    current_stock: int
    event_type: str  # sold, restocked, returned
    timestamp: str

@dataclass
class CustomerBehaviorEvent:
    customer_id: int
    event_type: str  # page_view, add_to_cart, purchase, search
    product_id: str | None
    session_id: str
    timestamp: str

class EventProducer:
    TOPIC_MAP = {
        OrderEvent: "orders",
        InventoryEvent: "inventory",
        CustomerBehaviorEvent: "customer-behavior",
    }

    PARTITION_KEY_MAP = {
        OrderEvent: lambda e: str(e.customer_id),
        InventoryEvent: lambda e: e.product_id,
        CustomerBehaviorEvent: lambda e: str(e.customer_id),
    }

    def __init__(self, bootstrap_servers: str):
        from confluent_kafka import Producer
        self.producer = Producer({
            "bootstrap.servers": bootstrap_servers,
            "acks": "all",
        })

    def send(self, event):
        event_type = type(event)
        topic = self.TOPIC_MAP[event_type]
        key = self.PARTITION_KEY_MAP[event_type](event)

        self.producer.produce(
            topic=topic,
            key=key.encode(),
            value=json.dumps(asdict(event)).encode(),
        )
        self.producer.flush()

from collections import defaultdict
from datetime import datetime, timedelta
import statistics

class RealTimeAggregator:
    def __init__(self, window_minutes: int = 5):
        self.window_size = timedelta(minutes=window_minutes)
        self.windows = defaultdict(lambda: {
            "order_count": 0,
            "total_revenue": 0,
            "by_region": defaultdict(float),
            "customers": set(),
        })
        self.historical_counts = []

    def _window_key(self, ts: datetime) -> str:
        seconds = ts.timestamp()
        window_start = seconds - (seconds % self.window_size.total_seconds())
        return datetime.fromtimestamp(window_start).isoformat()

    def process(self, event: dict) -> dict | None:
        ts = datetime.fromisoformat(event["timestamp"])
        key = self._window_key(ts)
        w = self.windows[key]

        w["order_count"] += 1
        w["total_revenue"] += event["total_amount"]
        w["by_region"][event["region"]] += event["total_amount"]
        w["customers"].add(event["customer_id"])

        # 異常検知
        alert = self._check_anomaly(key, w["order_count"])
        if alert:
            return alert
        return None

    def _check_anomaly(self, window_key: str, current_count: int) -> dict | None:
        if len(self.historical_counts) < 5:
            return None

        avg = statistics.mean(self.historical_counts[-20:])
        if current_count > avg * 3:
            return {
                "type": "ORDER_SPIKE",
                "window": window_key,
                "current": current_count,
                "average": round(avg, 1),
                "ratio": round(current_count / avg, 1),
            }
        return None

    def close_window(self, window_key: str) -> dict:
        w = self.windows[window_key]
        self.historical_counts.append(w["order_count"])

        region_ranking = sorted(
            w["by_region"].items(), key=lambda x: x[1], reverse=True
        )

        result = {
            "window": window_key,
            "order_count": w["order_count"],
            "total_revenue": w["total_revenue"],
            "unique_customers": len(w["customers"]),
            "avg_order_value": (
                w["total_revenue"] / w["order_count"]
                if w["order_count"] > 0 else 0
            ),
            "top_regions": region_ranking[:5],
        }

        del self.windows[window_key]
        return result

アーキテクチャ選択: Lambda（段階的にKappaへ移行）

理由:

• 既存のバッチパイプライン（Airflow + dbt）を活かす
• リアルタイム機能を段階的に追加
• 正確な日次レポートはバッチで担保

トピック設計:

Consumer Group設計:

障害時の対応:

• Consumer障害: Consumer Groupのリバランスで自動復旧
• Broker障害: レプリケーションによるLeader切り替え
• 処理エラー: DLQに退避 + アラート発報