分散システム完全ガイド: CAP定理から分散ML学習、Kafkaまで

┌─────────────────────────────────────────────┐
│              CAP Triangle                   │
│                                             │
│           Consistency (C)                  │
│               △                             │
│              / \                            │
│             /   \                           │
│            /     \                          │
│           /  CA   \                         │
│          /  (RDBMS)\                        │
│         /───────────\                       │
│        /  CP  │  AP  \                      │
│       / (HBase│(Cassandra)                  │
│      ▽────────┼────────▽                    │
│  Availability  Partition Tolerance          │
│     (A)              (P)                    │
└─────────────────────────────────────────────┘

P → A or C  (パーティション発生時)
E → L or C  (通常運用時)

Cassandra: PA/EL  (可用性 + 低レイテンシ優先)
etcd/Raft: PC/EC  (一貫性優先)

線形化可能性 (Linearizability)
  └── 最も強力。全操作が即座に反映される
  └── 例: etcd, Zookeeper

逐次一貫性 (Sequential Consistency)
  └── 全ノードが同じ順序で操作を観察する
  └── 例: CPUメモリモデル

因果一貫性 (Causal Consistency)
  └── 因果関係のある操作は順序が保証される
  └── 例: MongoDBの因果一貫性セッション

結果整合性 (Eventual Consistency)
  └── いつかは一貫性が保たれる
  └── 例: Cassandra, DNS, S3

┌──────────┐  timeout   ┌──────────────┐  majority   ┌────────┐
│          │──────────▶│   Candidate  │────────────▶│        │
│ Follower │            └──────────────┘             │ Leader │
│          │◀─────────────────────────────────────── │        │
└──────────┘          receive heartbeat              └────────┘
     ▲                                                    │
     └────────────────── heartbeat ──────────────────────┘

# etcdを使った分散ロックの実装
import etcd3
import time

def distributed_lock(etcd_client, lock_name, ttl=10):
    """Raftベースのetcdで分散ロックを実装"""
    lease = etcd_client.lease(ttl)
    lock_key = f"/locks/{lock_name}"

    # アトミックなCAS (Compare-And-Swap)
    success, _ = etcd_client.transaction(
        compare=[
            etcd3.transactions.create(lock_key, '==', 0)
        ],
        success=[
            etcd3.transactions.put(lock_key, 'locked', lease=lease)
        ],
        failure=[]
    )

    if success:
        print(f"ロック取得: {lock_name}")
        return lease
    else:
        print(f"ロック保持中: {lock_name}")
        return None

def release_lock(etcd_client, lease):
    """ロックの解放"""
    if lease:
        lease.revoke()
        print("ロック解放")

# 使用例
client = etcd3.client(host='localhost', port=2379)

lease = distributed_lock(client, 'my-resource')
if lease:
    try:
        # クリティカルセクションでの処理
        time.sleep(5)
    finally:
        release_lock(client, lease)

┌─────────────────────────────────────────────────────┐
│                  etcd Cluster (3 nodes)             │
│                                                     │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐      │
│  │  Leader  │───▶│Follower 1│    │Follower 2│      │
│  │  node-1  │───▶│  node-2  │    │  node-3  │      │
│  └──────────┘    └──────────┘    └──────────┘      │
│       ▲               │               │             │
│       └───────────────┴───────────────┘             │
│                  Raft Consensus                     │
└─────────────────────────────────────────────────────┘

フェーズ1 (Prepare):
  Coordinator ──▶ Participant A: "準備できた?"
  Coordinator ──▶ Participant B: "準備できた?"
  Participant A ──▶ Coordinator: "YES"
  Participant B ──▶ Coordinator: "YES"

フェーズ2 (Commit):
  Coordinator ──▶ Participant A: "コミットして!"
  Coordinator ──▶ Participant B: "コミットして!"

注文Saga:
  1. 注文作成      → 補償: 注文キャンセル
  2. 在庫引き落とし → 補償: 在庫復元
  3. 支払い処理    → 補償: 支払い払い戻し
  4. 配送開始      → 補償: 配送キャンセル

from typing import Callable, List
from dataclasses import dataclass
import logging

logger = logging.getLogger(__name__)

@dataclass
class SagaStep:
    name: str
    action: Callable
    compensating_action: Callable

class SagaOrchestrator:
    """Sagaオーケストレーターパターンの実装"""

    def __init__(self, steps: List[SagaStep]):
        self.steps = steps
        self.completed_steps = []

    def execute(self, context: dict) -> bool:
        for step in self.steps:
            try:
                logger.info(f"ステップ実行: {step.name}")
                step.action(context)
                self.completed_steps.append(step)
                logger.info(f"ステップ完了: {step.name}")
            except Exception as e:
                logger.error(f"ステップ失敗: {step.name}, エラー: {e}")
                self._compensate(context)
                return False
        return True

    def _compensate(self, context: dict):
        """逆順で補償トランザクションを実行"""
        logger.info("補償処理を開始...")
        for step in reversed(self.completed_steps):
            try:
                logger.info(f"補償実行: {step.name}")
                step.compensating_action(context)
            except Exception as e:
                logger.error(f"補償失敗: {step.name}, エラー: {e}")
                raise RuntimeError(f"Saga補償が{step.name}で失敗")

# 注文Sagaの定義
def create_order(ctx):
    ctx['order_id'] = 'ORD-001'
    print(f"注文作成: {ctx['order_id']}")

def cancel_order(ctx):
    print(f"注文キャンセル: {ctx['order_id']}")

def deduct_inventory(ctx):
    ctx['inventory_reserved'] = True
    print("在庫引き落とし完了")

def restore_inventory(ctx):
    print("在庫復元完了")

def process_payment(ctx):
    raise Exception("決済ゲートウェイタイムアウト")

def refund_payment(ctx):
    print("支払い払い戻し完了")

order_saga = SagaOrchestrator([
    SagaStep("create_order", create_order, cancel_order),
    SagaStep("deduct_inventory", deduct_inventory, restore_inventory),
    SagaStep("process_payment", process_payment, refund_payment),
])

context = {"user_id": "user-123", "amount": 50000}
result = order_saga.execute(context)
print(f"Saga結果: {'成功' if result else '補償して失敗'}")

CQRSアーキテクチャ:

  コマンド側 (書き込み)         クエリ側 (読み取り)
  ─────────────────            ────────────────
  POST /orders          ──▶    GET /orders/123
  PUT  /inventory       ──▶    GET /inventory/stats
       │                              ▲
       ▼                              │
  Event Store ──── Projection ────────┘
  (追記専用)       (読み取りモデル更新)

┌─────────────────────────────────────────────────────────────┐
│                     Kafka Cluster                           │
│                                                             │
│  Topic: orders (4 partitions, replication factor: 3)        │
│                                                             │
│  Broker 1      Broker 2      Broker 3                       │
│  ┌──────┐      ┌──────┐      ┌──────┐                       │
│  │ P0 L │      │ P0 F │      │ P0 F │  L=Leader, F=Follower │
│  │ P1 F │      │ P1 L │      │ P1 F │                       │
│  │ P2 F │      │ P2 F │      │ P2 L │                       │
│  │ P3 L │      │ P3 F │      │ P3 F │                       │
│  └──────┘      └──────┘      └──────┘                       │
│                                                             │
│  Consumer Group A          Consumer Group B                 │
│  Consumer 1: P0, P1        Consumer X: P0, P1, P2, P3      │
│  Consumer 2: P2, P3                                         │
└─────────────────────────────────────────────────────────────┘

from confluent_kafka import Producer, Consumer, KafkaError
import json
import time

# Producer設定
producer_config = {
    'bootstrap.servers': 'localhost:9092',
    'acks': 'all',           # 全ISRレプリカの確認後にack
    'retries': 3,
    'enable.idempotence': True,   # 厳密に1回の送信保証
    'compression.type': 'snappy',
}

def delivery_report(err, msg):
    if err is not None:
        print(f'メッセージ配信失敗: {err}')
    else:
        print(f'メッセージ配信完了: {msg.topic()} [{msg.partition()}] @ {msg.offset()}')

producer = Producer(producer_config)

# 注文イベントの発行
for i in range(10):
    order_event = {
        'order_id': f'ORD-{i:04d}',
        'user_id': f'user-{i % 5}',
        'amount': 10000 * (i + 1),
        'timestamp': time.time()
    }
    producer.produce(
        topic='orders',
        key=order_event['user_id'],   # 同じユーザーは同じパーティションへ
        value=json.dumps(order_event),
        callback=delivery_report
    )
    producer.poll(0)

producer.flush()
print("全メッセージ送信完了")

# Consumer設定
consumer_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processor-group',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,   # at-least-once保証のための手動コミット
}

consumer = Consumer(consumer_config)
consumer.subscribe(['orders'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            else:
                raise KafkaError(msg.error())

        event = json.loads(msg.value().decode('utf-8'))
        print(f"注文処理: {event['order_id']}, 金額: {event['amount']}")

        # 処理完了後にオフセットをコミット
        consumer.commit(asynchronous=False)

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

# Kafka Topic設定 (Strimzi Operator)
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: orders
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 12 # 最大並列コンシューマー数 = パーティション数
  replicas: 3 # レプリケーションファクター
  config:
    retention.ms: '604800000' # 7日間保持
    min.insync.replicas: '2' # 最低2つのISRレプリカ確認後にack
    compression.type: snappy
    cleanup.policy: delete

import hashlib
from sortedcontainers import SortedDict

class ConsistentHashRing:
    """コンシステントハッシングリングの実装"""

    def __init__(self, virtual_nodes=150):
        self.virtual_nodes = virtual_nodes
        self.ring = SortedDict()
        self.nodes = set()

    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_node(self, node: str):
        """仮想ノードを含めてリングにノードを追加"""
        self.nodes.add(node)
        for i in range(self.virtual_nodes):
            virtual_key = f"{node}#{i}"
            hash_val = self._hash(virtual_key)
            self.ring[hash_val] = node
        print(f"ノード追加: {node} ({self.virtual_nodes}個の仮想ノード)")

    def remove_node(self, node: str):
        """リングからノードを削除"""
        self.nodes.discard(node)
        keys_to_remove = [k for k, v in self.ring.items() if v == node]
        for k in keys_to_remove:
            del self.ring[k]
        print(f"ノード削除: {node}")

    def get_node(self, key: str) -> str:
        """キーに対応するノードを返す"""
        if not self.ring:
            raise Exception("リングが空です")

        hash_val = self._hash(key)
        idx = self.ring.bisect_left(hash_val)
        if idx == len(self.ring):
            idx = 0  # リングなのでラップアラウンド

        return self.ring.values()[idx]

    def get_nodes_for_replication(self, key: str, n: int) -> list:
        """レプリケーション用にn個のノードを返す"""
        if len(self.nodes) < n:
            raise Exception(f"ノード数不足: {len(self.nodes)} < {n}")

        hash_val = self._hash(key)
        idx = self.ring.bisect_left(hash_val)

        result_nodes = []
        seen_nodes = set()
        total = len(self.ring)

        for i in range(total):
            curr_idx = (idx + i) % total
            node = self.ring.values()[curr_idx]
            if node not in seen_nodes:
                seen_nodes.add(node)
                result_nodes.append(node)
                if len(result_nodes) == n:
                    break

        return result_nodes

# 使用例
ring = ConsistentHashRing(virtual_nodes=150)
for i in range(5):
    ring.add_node(f"cassandra-node-{i}")

for key in ["user:1001", "order:5555", "product:abc", "session:xyz"]:
    node = ring.get_node(key)
    replicas = ring.get_nodes_for_replication(key, 3)
    print(f"キー '{key}' -> プライマリ: {node}, レプリカ: {replicas}")

┌───────────────────────────────────────────────────────────────────┐
│                   Cassandraリング (6ノード)                        │
│                                                                   │
│                        Node 1                                     │
│                       /      \                                    │
│                   Node 6      Node 2                              │
│                     |            |                                │
│                   Node 5      Node 3                              │
│                       \      /                                    │
│                        Node 4                                     │
│                                                                   │
│  書き込みパス:                                                      │
│  Client → Coordinator → パーティションキーのハッシュ → 担当ノード   │
│                       → CommitLog (WAL) → Memtable → SSTable     │
│                                                                   │
│  読み取りパス:                                                      │
│  Client → Coordinator → 担当ノード → Row Cache / Bloomフィルタ     │
│                       → SSTableのマージ → 最新データを返す         │
└───────────────────────────────────────────────────────────────────┘

Parameter Server方式:
  ─────────────────────
  Worker 1 ─┐
  Worker 2 ─┤── PS (Parameter Server) ──▶ 勾配集計後に配布
  Worker 3 ─┘

  問題: PSがボトルネック。帯域幅の要求がO(n)で増加

Ring-AllReduce方式:
  ────────────────────
  Worker 1 ──▶ Worker 2 ──▶ Worker 3
     ▲                          │
     └──────────────────────────┘

  各Workerは隣接Workerとのみ通信。
  通信量はノード数に関わらず 2*(n-1)/n * data_size で一定

import os
import torch
import torch.nn as nn
import torch.optim as optim
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data import DataLoader, DistributedSampler

def setup(rank, world_size):
    """分散学習の初期化"""
    os.environ['MASTER_ADDR'] = 'localhost'
    os.environ['MASTER_PORT'] = '12355'

    # NCCLバックエンド: GPU間高速通信 (NVLink, InfiniBandを活用)
    dist.init_process_group(
        backend='nccl',
        rank=rank,
        world_size=world_size
    )
    torch.cuda.set_device(rank)

def cleanup():
    dist.destroy_process_group()

class SimpleModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(1024, 2048),
            nn.ReLU(),
            nn.Linear(2048, 1024),
            nn.ReLU(),
            nn.Linear(1024, 10)
        )

    def forward(self, x):
        return self.layers(x)

def train(rank, world_size, epochs=5):
    setup(rank, world_size)

    # 各GPUにモデルを配置してDDPでラップ
    model = SimpleModel().to(rank)
    ddp_model = DDP(model, device_ids=[rank])

    # DistributedSampler: 各プロセスが異なるデータを処理
    dataset = torch.utils.data.TensorDataset(
        torch.randn(1000, 1024),
        torch.randint(0, 10, (1000,))
    )
    sampler = DistributedSampler(dataset, num_replicas=world_size, rank=rank)
    dataloader = DataLoader(dataset, batch_size=32, sampler=sampler)

    optimizer = optim.Adam(ddp_model.parameters(), lr=1e-3)
    criterion = nn.CrossEntropyLoss()

    for epoch in range(epochs):
        sampler.set_epoch(epoch)  # エポックごとにシャッフルを再設定
        total_loss = 0.0

        for batch_x, batch_y in dataloader:
            batch_x = batch_x.to(rank)
            batch_y = batch_y.to(rank)

            optimizer.zero_grad()
            output = ddp_model(batch_x)
            loss = criterion(output, batch_y)
            loss.backward()  # 勾配のAllReduceが自動的に実行される
            optimizer.step()
            total_loss += loss.item()

        if rank == 0:  # メインプロセスのみログ出力
            print(f"Epoch {epoch}: Loss = {total_loss/len(dataloader):.4f}")

    cleanup()

# 実行: torchrun --nproc_per_node=4 train.py
if __name__ == '__main__':
    world_size = torch.cuda.device_count()
    torch.multiprocessing.spawn(
        train,
        args=(world_size,),
        nprocs=world_size,
        join=True
    )

NCCL (NVIDIA Collective Communications Library):

  シングルノード (NVLink):
  GPU0 ←──── NVLink ────→ GPU1
    │                       │
   600 GB/s 双方向帯域幅

  マルチノード (InfiniBand + RDMA):
  Node 1 ←── IB (200 Gb/s) ──→ Node 2
    │                              │
  RDMA: CPUを介さずGPU間で直接通信

バックエンドの選択:
  - nccl: GPU分散学習 (推奨)
  - gloo: CPU学習またはデバッグ
  - mpi: HPC環境

リクエストフロー (Jaeger/Zipkinによるトレース):

  Browser → API GW → Order Svc → Inventory Svc → DB
    │           │          │            │          │
    │  TraceID: abc123     │            │          │
    │  SpanID: 001    SpanID: 002  SpanID: 003     │
    │                      │            │          │
    └──────────────────────┴────────────┴──────────┘
             合計レイテンシ: 120ms
             Order Svc: 80ms (ボトルネック!)

OpenTelemetryのレイヤー構成:
  - API: 計装コードの記述
  - SDK: テレメトリデータの処理とエクスポート
  - Exporter: Jaeger/Zipkin/OTLPエンドポイントへ送信

┌─────────────────────────────────────────────────────┐
│                   ELK Stack                         │
│                                                     │
│  App Logs ──▶ Filebeat ──▶ Logstash ──▶ Elasticsearch│
│  K8s Logs ──▶ Fluentd ──/             ──▶ Kibana    │
│                                         (可視化)    │
│                                                     │
│  モダンな代替スタック:                               │
│  App Logs ──▶ Promtail ──▶ Loki ──▶ Grafana         │
└─────────────────────────────────────────────────────┘

# Prometheusスクレイピング設定
scrape_configs:
  - job_name: 'kafka'
    static_configs:
      - targets: ['kafka-broker-1:9090', 'kafka-broker-2:9090']
    metrics_path: '/metrics'
    scrape_interval: 15s

  - job_name: 'pytorch-training'
    static_configs:
      - targets: ['trainer-0:8080', 'trainer-1:8080']