# 機能設計書 29-シャッフル

## 概要

本ドキュメントは、Apache Flinkのシャッフル機能（flink-runtime モジュール内ShuffleMaster）の設計仕様を記述する。シャッフル機能は、タスク間のデータ交換（パーティション転送）を担当する。

### 本機能の処理概要

ShuffleMasterは、Flinkジョブ内のオペレーター間でデータを転送するためのシャッフルサービスを管理する。ResultPartitionの登録とShuffleDescriptorの生成を行う。

**業務上の目的・背景**：分散処理において、keyByやrebalance等の操作でデータを異なるタスク間で転送する必要がある。効率的なデータ交換メカニズムは、ジョブ全体のスループットとレイテンシに直結する。

**機能の利用シーン**：
- keyBy操作によるキーベースのパーティショニング
- rebalance操作によるラウンドロビン配分
- broadcast操作による全パーティションへの配信
- forward操作による1対1転送

**主要な処理内容**：
1. ResultPartitionの登録
2. ShuffleDescriptorの生成
3. パーティションの解放
4. シャッフルメモリの計算
5. パーティションメトリクスの取得

**関連システム・外部連携**：
- JobMaster（パーティション管理）
- TaskExecutor（データ送受信）
- ShuffleEnvironment（ローカルシャッフル環境）

**権限による制御**：JobMaster内でのみアクセス可能。

## 関連画面

本機能は画面機能マッピングには直接含まれていない。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 内部処理として使用 |

## 機能種別

データ転送 / ネットワーク通信

## 入力仕様

### 入力パラメータ

#### パーティション登録

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| jobID | JobID | Yes | ジョブID | UUID形式 |
| partitionDescriptor | PartitionDescriptor | Yes | パーティション情報 | null不可 |
| producerDescriptor | ProducerDescriptor | Yes | プロデューサ情報 | null不可 |

### 入力データソース

- JobMasterからのパーティション登録要求
- TaskExecutorからのメトリクス情報

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| ShuffleDescriptor | ShuffleDescriptor | シャッフル記述子 |
| MemorySize | MemorySize | シャッフルメモリサイズ |
| PartitionWithMetrics | PartitionWithMetrics | パーティションとメトリクス |

### 出力先

- TaskExecutor（シャッフル記述子）
- SchedulerNG（パーティション情報）

## 処理フロー

### 処理シーケンス

```
1. ShuffleMaster初期化
   └─ ジョブ起動時に初期化
   └─ registerJob() でジョブ登録

2. パーティション登録
   └─ registerPartitionWithProducer() 呼び出し
   └─ ShuffleDescriptor生成

3. タスクデプロイ
   └─ ShuffleDescriptorをタスクに渡す
   └─ TaskExecutorでShuffleEnvironment初期化

4. データ交換
   └─ ResultPartitionからデータ書き込み
   └─ InputGateでデータ読み込み

5. パーティション解放
   └─ releasePartitionExternally() でリソース解放
```

### フローチャート

```mermaid
flowchart TD
    A[registerJob] --> B[パーティション登録要求]
    B --> C[registerPartitionWithProducer]
    C --> D[ShuffleDescriptor生成]

    D --> E[タスクデプロイ]
    E --> F[ShuffleEnvironment初期化]

    F --> G[データ交換開始]
    G --> H[ResultPartition書き込み]
    H --> I[ネットワーク転送]
    I --> J[InputGate読み込み]

    J --> K{パーティション完了?}
    K -->|No| H
    K -->|Yes| L[releasePartitionExternally]
    L --> M[リソース解放]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | パーティションID一意性 | ResultPartitionIDは一意 | パーティション作成時 |
| BR-002 | バッファプール | ネットワークバッファはプール化 | データ転送時 |
| BR-003 | バックプレッシャー | 下流が遅い場合は上流を制限 | データ転送時 |
| BR-004 | ローカル優先 | 同一TM内はローカル転送 | 転送経路決定時 |

### 計算ロジック

- シャッフルメモリサイズの計算
- パーティションサイズの推定

## データベース操作仕様

### 操作別データベース影響一覧

本機能は直接データベースを操作しない。

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | システムエラー | ネットワーク障害 | 再接続 |
| PartitionNotFoundException | ビジネスエラー | パーティション不在 | 再スケジュール |
| ProducerFailedException | 処理エラー | プロデューサ障害 | タスク再起動 |

### リトライ仕様

- ネットワーク障害: 再接続を試行
- パーティション不在: タスク再スケジュール

## トランザクション仕様

データ転送はチェックポイントと連携してExactly-once保証。

## パフォーマンス要件

- スループット: ネットワーク帯域に依存
- レイテンシ: ミリ秒〜数十ミリ秒

## セキュリティ考慮事項

- SSL/TLSによるデータ転送暗号化（オプション）

## 備考

- NettyShuffleMaster: Nettyベースの標準実装
- 外部シャッフルサービスの統合も可能

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: インターフェースを理解する

ShuffleMasterのインターフェースを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ShuffleMaster.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleMaster.java` | メインインターフェース |
| 1-2 | ShuffleDescriptor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java` | シャッフル記述子 |

**読解のコツ**:
- **39行目**: ShuffleMasterはジェネリクスでShuffleDescriptor型を指定
- **44-46行目**: start()/close()でライフサイクル管理
- **61-64行目**: registerJob()/unregisterJob()でジョブ管理
- **87-90行目**: registerPartitionWithProducer()でパーティション登録

#### Step 2: データ構造を理解する

関連するデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | PartitionDescriptor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/PartitionDescriptor.java` | パーティション記述 |
| 2-2 | ProducerDescriptor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ProducerDescriptor.java` | プロデューサ記述 |

### プログラム呼び出し階層図

```
ShuffleMaster (interface)
    │
    ├─ start()
    │      └─ シャッフルサービス初期化
    │
    ├─ registerJob()
    │      └─ JobShuffleContext登録
    │
    ├─ registerPartitionWithProducer()
    │      └─ ShuffleDescriptor生成
    │
    ├─ releasePartitionExternally()
    │      └─ 外部リソース解放
    │
    ├─ computeShuffleMemorySizeForTask()
    │      └─ メモリサイズ計算
    │
    └─ getPartitionWithMetrics()
           └─ パーティションメトリクス取得
```

### データフロー図

```
[入力]                    [処理]                           [出力]

パーティション登録 ────▶ ShuffleMaster ─────────────▶ ShuffleDescriptor
                              │
                              ▼
                     パーティション管理
                              │
                              ▼
                       [データ転送]
                              │
           ┌──────────────────┼──────────────────┐
           │                  │                  │
           ▼                  ▼                  ▼
    ResultPartition     ネットワーク        InputGate
    (送信側)             バッファ          (受信側)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ShuffleMaster.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleMaster.java` | ソース | メインインターフェース |
| ShuffleDescriptor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java` | ソース | シャッフル記述子 |
| PartitionDescriptor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/PartitionDescriptor.java` | ソース | パーティション記述 |
| ProducerDescriptor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ProducerDescriptor.java` | ソース | プロデューサ記述 |
| JobShuffleContext.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/JobShuffleContext.java` | ソース | ジョブコンテキスト |
| ShuffleEnvironment.java | `flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleEnvironment.java` | ソース | ローカル環境 |
