# 機能設計書 10-アキュムレータ

## 概要

本ドキュメントは、Apache Sparkのアキュムレータ機能の設計を記述する。アキュムレータは分散環境で安全に加算可能な共有変数を提供し、カウンターや合計値の集約に使用する。

### 本機能の処理概要

AccumulatorV2は、分散タスク実行中に安全に値を集約するための共有変数メカニズムである。各タスクのコピーに対して加算操作を行い、ドライバー上で全タスクの結果をマージすることで、グローバルな集約結果を得る。

**業務上の目的・背景**：分散処理において、全タスクを横断したカウンターや集計値が必要な場面がある（例：処理レコード数のカウント、エラー件数の集計、カスタム集約値の計算）。通常の共有変数は分散環境では安全に更新できないため、アキュムレータが「加算のみ」の安全な共有変数メカニズムを提供する。また、Spark内部ではタスクメトリクス（実行時間、GC時間、シャッフルバイト数等）の収集にも使用される。

**機能の利用シーン**：タスク処理中のカウンターインクリメント、カスタム集約値の計算、内部メトリクス収集。ドライバーでのみ値を読み取り可能。

**主要な処理内容**：
1. アキュムレータの登録（SparkContext.register）
2. タスク実行時のローカルコピー作成とアキュムレータ更新
3. タスク完了時のアキュムレータ値送信（ハートビートまたはタスク結果に含む）
4. ドライバー上でのマージ処理
5. InternalAccumulatorによる内部メトリクス収集
6. ContextCleanerによる不要アキュムレータのクリーンアップ

**関連システム・外部連携**：SparkContext（登録API）、TaskContext（タスク内アクセス）、ハートビート（メトリクス送信）

**権限による制御**：アキュムレータの値はドライバーでのみ読み取り可能。Executor上ではadd操作のみ許可される。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 該当なし | - | - | アキュムレータの値はWeb UIのStage Detailページ等でタスクメトリクスとして表示される（InternalAccumulator経由） |

## 機能種別

分散集約 / メトリクス収集 / 共有変数管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| sc | SparkContext | Yes | 登録先SparkContext | null不可 |
| name | Option[String] | No | アキュムレータの名前 | - |
| countFailedValues | Boolean | No | 失敗タスクの値も集約するか | デフォルトfalse |
| initialValue | IN | Yes | 初期値（resetで使用） | - |
| value | IN | Yes | 加算する値 | - |

### 入力データソース

- SparkContext.register(): アキュムレータの登録
- AccumulatorV2.add(): タスク内での値の加算
- InternalAccumulator: 内部メトリクス名定義

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| value | OUT | アキュムレータの現在の集約値（ドライバー側のみ） |
| AccumulableInfo | AccumulableInfo | アキュムレータのメタ情報（ID、名前、値） |

### 出力先

ドライバー上のAccumulatorContextグローバルマップ。タスク結果またはハートビートに含めてドライバーへ送信。

## 処理フロー

### 処理シーケンス

```
1. アキュムレータ生成・登録
   └─ AccumulatorV2サブクラスのインスタンス生成、SparkContext.registerで登録
2. AccumulatorContextへの登録
   └─ グローバルIDの割り当てとマップへの登録
3. タスクシリアライズ時のコピー
   └─ タスクデータにアキュムレータ参照を含めてシリアライズ
4. Executor上でのadd操作
   └─ タスク実行中にadd()メソッドでローカルコピーに値を加算
5. タスク完了時の値送信
   └─ タスク結果にアキュムレータ更新値を含めてドライバーへ送信
6. ドライバー上でのmerge
   └─ 受信したタスクのアキュムレータ値を元のアキュムレータにmerge
7. ハートビート経由の中間報告（オプション）
   └─ excludeFromHeartbeat=falseのアキュムレータはハートビートに含まれる
```

### フローチャート

```mermaid
flowchart TD
    A[AccumulatorV2生成] --> B[SparkContext.register]
    B --> C[AccumulatorContext登録]
    C --> D[タスクシリアライズ]
    D --> E[Executor上でタスク実行]
    E --> F[add: ローカルコピーに値を加算]
    F --> G{タスク成功?}
    G -->|Yes| H[タスク結果にアキュムレータ値を含む]
    G -->|No| I{countFailedValues=true?}
    I -->|Yes| H
    I -->|No| J[値を破棄]
    H --> K[ドライバーでmerge]
    K --> L[集約値更新]
    L --> M[value()で結果取得]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | 加算のみ | アキュムレータはadd操作のみサポート（減算不可） | 常時 |
| BR-02 | ドライバー読み取り | valueはドライバー上でのみ読み取り可能 | 常時 |
| BR-03 | 失敗タスク制御 | countFailedValues=falseの場合、失敗タスクのアキュムレータ値は破棄 | タスク失敗時 |
| BR-04 | 二重登録禁止 | 同一アキュムレータの二度目のregisterはIllegalStateException | register時 |
| BR-05 | 内部メトリクス | InternalAccumulatorはcountFailedValues=trueで常に集約 | 内部メトリクス |

### 計算ロジック

アキュムレータのマージ: `driver.merge(taskCopy)` が呼ばれ、各タスクのローカルコピーの値がドライバーのアキュムレータにマージされる。マージ操作は可換・結合的であることが前提。

組み込みアキュムレータ:
- LongAccumulator: Long値の合計
- DoubleAccumulator: Double値の合計
- CollectionAccumulator: コレクションへの要素追加

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | アキュムレータはデータベース操作を行わない |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalStateException | 二重登録 | 同一アキュムレータを二度registerした場合 | 新しいインスタンスを生成してregister |
| IllegalStateException | メタデータ未初期化 | register前にid/nameにアクセス | 先にregisterを行う |

### リトライ仕様

タスク失敗時のリトライではアキュムレータのローカルコピーはリセットされる。countFailedValues=trueの内部メトリクスのみ失敗値が保持される。

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

アキュムレータの値はアトミックに読み書きできる型（Long, Double等）またはスレッドセーフなコレクションを使用することが推奨される。これはOUT型がドライバー上の他スレッドから読み取られうるためである。

## パフォーマンス要件

- add操作: ナノ秒オーダー（ローカル操作）
- merge操作: マイクロ秒オーダー（ドライバー上）
- ハートビート経由のメトリクス送信: spark.executor.heartbeatInterval間隔

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

- アキュムレータの値はタスク結果に含まれてドライバーに送信される
- spark.authenticate有効時は通信が認証・暗号化される
- ユーザー定義アキュムレータの値はシリアライズ可能である必要がある

## 備考

- AccumulatorV2はSpark 2.0で導入された新APIであり、旧Accumulator APIを置き換えた
- InternalAccumulatorはタスクメトリクス（executorRunTime, jvmGCTime, shuffleRead/Write等）の収集に使用される
- AccumulatorContextはWeakReferenceでアキュムレータを保持し、GCによる自動クリーンアップを支援する

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | AccumulatorV2.scala | `core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala` | 抽象クラスAccumulatorV2の定義。IN型（入力）とOUT型（出力）のジェネリクス |
| 1-2 | InternalAccumulator.scala | `core/src/main/scala/org/apache/spark/InternalAccumulator.scala` | 内部メトリクス用アキュムレータ名の定義 |

**読解のコツ**: AccumulatorV2.scalaのクラス定義（44行目）で`IN`と`OUT`の2つの型パラメータがある点に注意。`add(IN)`で値を加算し、`value: OUT`で結果を取得する。`isZero`, `copy`, `reset`, `add`, `merge`が実装必須の抽象メソッド。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AccumulatorV2.scala | `core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala` | register(), add(), merge(), value()メソッド |
| 2-2 | SparkContext.scala | `core/src/main/scala/org/apache/spark/SparkContext.scala` | register()メソッド（ユーザーAPI） |

**主要処理フロー**:
- **44行目**: AccumulatorV2抽象クラス定義
- **50-59行目**: register()メソッド - AccumulatorContextへの登録
- **67-68行目**: isRegistered判定
- **79-82行目**: id取得
- **87-95行目**: name取得
- **98-100行目**: countFailedValues説明

#### Step 3: 内部メトリクスの定義を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | InternalAccumulator.scala | `core/src/main/scala/org/apache/spark/InternalAccumulator.scala` | メトリクス名プレフィックス定義 |

**主要処理フロー**:
- **24行目**: InternalAccumulatorオブジェクト
- **26-31行目**: メトリクスプレフィックス（METRICS_PREFIX, SHUFFLE_READ等）
- **34-47行目**: タスクレベルメトリクス名（executorRunTime, jvmGCTime等）
- **52-70行目**: シャッフル読み取りメトリクス名
- **73-77行目**: シャッフル書き込みメトリクス名
- **80-83行目**: 出力メトリクス名
- **86-89行目**: 入力メトリクス名

#### Step 4: AccumulatorContextを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | AccumulatorV2.scala（AccumulatorContextオブジェクト） | `core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala` | グローバルID管理とWeakReferenceマップ |

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

```
SparkContext.register(accumulator)
    |
    +-- AccumulatorV2.register(sc, name, countFailedValues)
            +-- AccumulatorContext.newId() [一意ID生成]
            +-- AccumulatorContext.register(this) [グローバルマップ登録]
            +-- ContextCleaner.registerAccumulatorForCleanup()

[タスク実行時]
Executor.TaskRunner.run()
    |
    +-- タスクデシリアライズ（アキュムレータコピー含む）
    +-- task.run()
    |       +-- AccumulatorV2.add(value) [ローカルコピーに加算]
    +-- タスク結果にアキュムレータ更新を含める
    +-- ドライバーへ結果送信

[ドライバー側]
DAGScheduler.handleTaskCompletion()
    |
    +-- Accumulators.merge(taskAccumulators)
            +-- AccumulatorV2.merge(other) [各アキュムレータのmerge]
```

### データフロー図

```
[ドライバー]                 [処理]                         [Executor]

AccumulatorV2.register ───▶ AccumulatorContext
                              (ID生成・マップ登録)

                            タスクシリアライズ      ───▶  ローカルコピー生成
                                                          |
                                                          +-- add(value)
                                                          |   [ローカル加算]
                                                          |
                                                   ◀───  タスク結果送信
                                                          (アキュムレータ値含む)

AccumulatorV2.merge    ◀── タスク完了処理
  (集約値更新)

AccumulatorV2.value    ───▶ 集約結果取得
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| AccumulatorV2.scala | `core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala` | ソース | アキュムレータ基底クラスとAccumulatorContext |
| InternalAccumulator.scala | `core/src/main/scala/org/apache/spark/InternalAccumulator.scala` | ソース | 内部メトリクス名定義 |
| AccumulableInfo.scala | `core/src/main/scala/org/apache/spark/scheduler/AccumulableInfo.scala` | ソース | アキュムレータ情報のシリアライズ用 |
| TaskMetrics.scala | `core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala` | ソース | タスクメトリクス（内部アキュムレータ使用） |
| SparkContext.scala | `core/src/main/scala/org/apache/spark/SparkContext.scala` | ソース | register API |
| ContextCleaner.scala | `core/src/main/scala/org/apache/spark/ContextCleaner.scala` | ソース | アキュムレータクリーンアップ |
| DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | ソース | タスク完了時のアキュムレータマージ |
