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

## 概要

本ドキュメントは、Apache Sparkのシャッフル機構の設計を記述する。シャッフル機構はMapReduceスタイルのデータ再分配を効率的に実行し、ソートベースのシャッフルマネージャによりデータの書き込み・読み取りを管理する。

### 本機能の処理概要

SortShuffleManagerは、マップ出力をパーティションIDに基づきソートし、単一のマップ出力ファイルに書き込む。リデューサーはこのファイルの連続領域を読み取って自分のパーティションのデータを取得する。

**業務上の目的・背景**：groupByKey、reduceByKey、join等のwide dependency操作では、データをクラスタ全体で再分配（シャッフル）する必要がある。シャッフルはSparkの性能ボトルネックとなりやすいため、効率的なソートベースシャッフル機構により、メモリ使用量の最適化、ディスクI/Oの削減、ネットワーク転送の効率化を実現する。

**機能の利用シーン**：RDDのgroupByKey、reduceByKey、join、repartition等のwide dependency変換操作実行時に自動的に使用される。

**主要な処理内容**：
1. シャッフルの登録（registerShuffle）
2. マップ出力の書き込み（ShuffleWriter）
3. シリアライズドソートモードとデシリアライズドソートモードの選択
4. メモリ不足時のディスクスピルとマージ
5. インデックスファイルによるパーティションオフセット管理
6. リデューサーによるマップ出力の読み取り（ShuffleReader/BlockStoreShuffleReader）
7. Push-based Shuffle（シャッフルプッシュ最適化）

**関連システム・外部連携**：BlockManager（データ保存）、MapOutputTracker（マップ出力位置追跡）、External Shuffle Service

**権限による制御**：シャッフル機構は内部コンポーネントであり、権限制御は持たない。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 4 | Stage Detail（ステージ詳細） | 補助機能 | シャッフルRead/Writeのバイト数・レコード数・待ち時間等のメトリクスを表示 |

## 機能種別

データ再分配 / ソート・マージ / I/O最適化

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| shuffleDep | ShuffleDependency | Yes | シャッフル依存関係定義 | null不可 |
| mapId | Long | Yes | マップタスクID | 0以上 |
| context | TaskContext | Yes | タスクコンテキスト | null不可 |
| records | Iterator[Product2[K, V]] | Yes | マップ出力レコード | - |

### 入力データソース

マップタスクの計算結果（キー・バリューペアのイテレータ）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| マップ出力ファイル | File | ソート済みマップ出力データ |
| インデックスファイル | File | パーティションオフセット情報 |
| MapStatus | MapStatus | マップ出力の場所とサイズ情報 |
| ShuffleReadMetrics | ShuffleReadMetrics | シャッフル読み取りメトリクス |
| ShuffleWriteMetrics | ShuffleWriteMetrics | シャッフル書き込みメトリクス |

### 出力先

ローカルディスク（マップ出力ファイル）、MapOutputTracker（マップ出力位置情報）

## 処理フロー

### 処理シーケンス

```
1. シャッフル登録（registerShuffle）
   └─ ShuffleHandle（BaseShuffleHandle等）の生成
2. ShuffleWriterの取得
   └─ 条件に基づきUnsafeShuffleWriter or SortShuffleWriter選択
3. マップ出力書き込み
   └─ レコードをパーティションIDでソートし、単一ファイルに書き込み
4. スピル処理（メモリ不足時）
   └─ メモリ上のソートデータをディスクにスピル
5. マージ処理
   └─ 複数のスピルファイルを最終出力ファイルにマージ
6. インデックスファイル書き込み
   └─ 各パーティションのオフセットをインデックスファイルに記録
7. リデューサー読み取り
   └─ BlockStoreShuffleReaderがインデックスに基づきデータ取得
```

### フローチャート

```mermaid
flowchart TD
    A[シャッフル開始] --> B{シリアライズドソート条件?}
    B -->|Yes| C[UnsafeShuffleWriter]
    B -->|No| D[SortShuffleWriter]
    C --> E[シリアライズ済みバイナリソート]
    D --> F[デシリアライズ済みオブジェクトソート]
    E --> G{メモリ不足?}
    F --> G
    G -->|Yes| H[ディスクスピル]
    G -->|No| I[最終出力ファイル書き込み]
    H --> I
    I --> J[インデックスファイル書き込み]
    J --> K[MapStatus返却]
    K --> L[BlockStoreShuffleReader読み取り]
    L --> M[パーティションデータ返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | シリアライズドソート条件 | map-side combineなし、シリアライザがリロケーション対応、パーティション数16777216以下 | 全3条件満たす場合 |
| BR-02 | 単一ファイル出力 | 各マップタスクは1つのデータファイルと1つのインデックスファイルを出力 | 常時 |
| BR-03 | スピル閾値 | メモリ使用量が閾値を超えた場合にディスクスピル | メモリ不足時 |
| BR-04 | チェックサム検証 | spark.shuffle.checksum.enabled=trueでチェックサム検証を実施 | 設定有効時 |

### 計算ロジック

シリアライズドソートモードの最適化:
- シリアライズ済みバイナリデータを直接ソート（デシリアライズ不要）
- 8バイト/レコードの圧縮ポインタでキャッシュ効率を向上
- スピルマージ時にシリアライズデータの直接連結（圧縮対応時）

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | シャッフル機構はファイルベースのI/Oを使用。データベースは使用しない。 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| FetchFailedException | フェッチ失敗 | マップ出力の読み取り失敗 | DAGSchedulerがマップステージを再実行 |
| SparkOutOfMemoryError | メモリエラー | ソート用メモリ不足 | ディスクスピル後に再試行 |
| IOException | I/Oエラー | ディスク書き込み/読み取り失敗 | タスク失敗→リトライ |

### リトライ仕様

マップ出力読み取りの失敗はDAGSchedulerに通知され、親のShuffleMapStageが再実行される。

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

シャッフル書き込みはタスク単位でアトミック。IndexShuffleBlockResolverがインデックスとデータファイルの一貫性を保証する。

## パフォーマンス要件

- シリアライズドソートモードでGCオーバーヘッド削減
- NIO transferToによる効率的なデータコピー（圧縮連結対応時）
- Push-based Shuffleによるリデューサー側I/O削減

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

- シャッフルデータはローカルディスクに平文で保存される
- リモートフェッチ時はspark.authenticate有効時に認証・暗号化される
- External Shuffle Service使用時はシャッフルデータがExecutor終了後も永続化

## 備考

- SortShuffleManagerはSpark 2.0以降唯一のシャッフル実装（HashShuffleManagerは削除済み）
- spark.shuffle.sort.bypassMergeThresholdによりバイパスマージソートが使用される場合がある
- Push-based Shuffle（マージ化シャッフル）はSpark 3.2で導入

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ShuffleHandle.scala | `core/src/main/scala/org/apache/spark/shuffle/ShuffleHandle.scala` | シャッフルハンドルの定義 |
| 1-2 | BaseShuffleHandle.scala | `core/src/main/scala/org/apache/spark/shuffle/BaseShuffleHandle.scala` | 基本シャッフルハンドル |
| 1-3 | MapStatus.scala | `core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala` | マップ出力の位置・サイズ情報 |

**読解のコツ**: SortShuffleManager.scalaの30-72行目のScaladocに、シリアライズドソートモードとデシリアライズドソートモードの違いが詳しく説明されている。特にシリアライズドソートの3条件を理解することが重要。

#### Step 2: エントリーポイントを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SortShuffleManager.scala | `core/src/main/scala/org/apache/spark/shuffle/sort/SortShuffleManager.scala` | registerShuffle(), getWriter(), getReader() |

**主要処理フロー**:
- **30-72行目**: Scaladocでシャッフル設計の詳細説明
- **73行目**: クラス定義
- registerShuffle(): ShuffleHandleの種類決定
- getWriter(): 条件に基づくWriter選択
- getReader(): BlockStoreShuffleReaderの生成

#### Step 3: 書き込みパスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SortShuffleWriter.scala | `core/src/main/scala/org/apache/spark/shuffle/sort/SortShuffleWriter.scala` | デシリアライズドソートの書き込み |
| 3-2 | UnsafeShuffleWriter.java | `core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java` | シリアライズドソートの書き込み |
| 3-3 | IndexShuffleBlockResolver.scala | `core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala` | インデックス/データファイル管理 |

#### Step 4: 読み取りパスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | BlockStoreShuffleReader.scala | `core/src/main/scala/org/apache/spark/shuffle/BlockStoreShuffleReader.scala` | シャッフルデータの読み取り |
| 4-2 | ShuffleBlockFetcherIterator.scala | `core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala` | リモートブロックの取得 |

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

```
ShuffleMapTask.runTask()
    |
    +-- SortShuffleManager.getWriter()
    |       +-- SortShuffleWriter / UnsafeShuffleWriter
    |               +-- ExternalSorter.insertAll() [ソート]
    |               +-- ExternalSorter.writePartitionedMapOutput() [書き込み]
    |               +-- IndexShuffleBlockResolver.writeIndexFileAndCommit()
    |
    +-- SortShuffleManager.getReader()
            +-- BlockStoreShuffleReader.read()
                    +-- ShuffleBlockFetcherIterator
                    |       +-- BlockManager.getLocalBytes()
                    |       +-- BlockManager.getRemoteBytes()
                    +-- InterruptibleIterator
```

### データフロー図

```
[Map側]                      [処理]                         [Reduce側]

レコード入力         ───▶  SortShuffleWriter      ───▶  マップ出力ファイル
(Iterator[K,V])            |                              + インデックスファイル
                           +-- パーティションIDソート
                           +-- スピル/マージ
                           +-- MapStatus → MapOutputTracker

                           MapOutputTracker        ───▶  BlockStoreShuffleReader
                           (ブロック位置情報)               |
                                                          +-- ShuffleBlockFetcherIterator
                                                          +-- パーティションデータ取得
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SortShuffleManager.scala | `core/src/main/scala/org/apache/spark/shuffle/sort/SortShuffleManager.scala` | ソース | シャッフルマネージャ |
| SortShuffleWriter.scala | `core/src/main/scala/org/apache/spark/shuffle/sort/SortShuffleWriter.scala` | ソース | デシリアライズドソートWriter |
| UnsafeShuffleWriter.java | `core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java` | ソース | シリアライズドソートWriter |
| BlockStoreShuffleReader.scala | `core/src/main/scala/org/apache/spark/shuffle/BlockStoreShuffleReader.scala` | ソース | シャッフルReader |
| IndexShuffleBlockResolver.scala | `core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala` | ソース | インデックスファイル管理 |
| ShuffleBlockFetcherIterator.scala | `core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala` | ソース | リモートブロック取得 |
| ShuffleBlockPusher.scala | `core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala` | ソース | Push-basedシャッフル |
| ShuffleManager.scala | `core/src/main/scala/org/apache/spark/shuffle/ShuffleManager.scala` | ソース | シャッフルマネージャインターフェース |
| ShuffleWriter.scala | `core/src/main/scala/org/apache/spark/shuffle/ShuffleWriter.scala` | ソース | Writer抽象クラス |
| ShuffleReader.scala | `core/src/main/scala/org/apache/spark/shuffle/ShuffleReader.scala` | ソース | Reader抽象クラス |
| ShuffleChecksumUtils.scala | `core/src/main/scala/org/apache/spark/shuffle/ShuffleChecksumUtils.scala` | ソース | チェックサムユーティリティ |
