# 機能設計書 79-Pregelフレームワーク

## 概要

本ドキュメントは、Apache Spark GraphXにおけるPregelフレームワークの設計を記述する。

### 本機能の処理概要

Pregelは、GoogleのPregel論文に基づくバルク同期並列（BSP）メッセージパッシングAPIの実装である。GraphXのPregelは、元のPregel APIとは異なり、メッセージ送信計算をエッジ上で分解（factor）し、両頂点の属性を読み取り可能にし、メッセージをグラフ構造に制約する。これにより、より効率的な分散実行と柔軟なグラフ計算が可能になる。

**業務上の目的・背景**：グラフアルゴリズム（PageRank、連結成分、最短経路、ラベル伝播など）の反復的メッセージパッシング計算を統一的なフレームワークで実装するための基盤。

**機能の利用シーン**：GraphXの全グラフアルゴリズムライブラリ（PageRank、ConnectedComponents、StronglyConnectedComponents、ShortestPaths、LabelPropagation）の反復計算エンジンとして使用される。ユーザー定義のカスタムグラフアルゴリズムの実装にも使用される。

**主要な処理内容**：
1. 初回: 全頂点にinitialMsgを送信しvprogを実行
2. エッジ上でsendMsgを実行し、メッセージを生成
3. mergeMsgでメッセージを頂点ごとに集約
4. メッセージを受信した頂点でvprogを再実行
5. メッセージがなくなるかmaxIterationsに達するまで反復
6. チェックポイントによるリネージ切断をサポート

**関連システム・外部連携**：GraphXのグラフ構造データ管理（No.71）に依存。GraphXUtils.mapReduceTripletsを内部で使用。

**権限による制御**：特に権限による制御は行われない。

## 関連画面

本機能に直接関連するUI画面はない。

## 機能種別

計算処理基盤 / 反復フレームワーク

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| graph | Graph[VD, ED] | Yes | 入力グラフ | - |
| initialMsg | A | Yes | 最初のスーパーステップで全頂点に送信されるメッセージ | - |
| maxIterations | Int | No | 最大反復回数（デフォルト: Int.MaxValue） | > 0 |
| activeDirection | EdgeDirection | No | メッセージ送信対象のエッジ方向（デフォルト: EdgeDirection.Either） | - |
| vprog | (VertexId, VD, A) => VD | Yes | 頂点プログラム | - |
| sendMsg | EdgeTriplet[VD, ED] => Iterator[(VertexId, A)] | Yes | メッセージ送信関数 | - |
| mergeMsg | (A, A) => A | Yes | メッセージ結合関数（可換・結合法則を満たすこと） | - |

### 入力データソース

Graph[VD, ED] - GraphXのグラフ構造。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Graph[VD, ED] | Graph | 反復計算後の結果グラフ（頂点属性が更新済み） |

### 出力先

メモリ上のGraph RDD

## 処理フロー

### 処理シーケンス

```
1. 初期化
   ├─ g = graph.mapVertices((vid, vdata) => vprog(vid, vdata, initialMsg))
   ├─ graphCheckpointer = new PeriodicGraphCheckpointer(checkpointInterval)
   ├─ messages = GraphXUtils.mapReduceTriplets(g, sendMsg, mergeMsg)
   └─ messageCheckpointer = new PeriodicRDDCheckpointer(checkpointInterval)

2. 反復ループ (isActiveMessagesNonEmpty && i < maxIterations)
   ├─ g = g.joinVertices(messages)(vprog) [メッセージ受信頂点のみvprogを実行]
   ├─ graphCheckpointer.update(g)
   ├─ messages = GraphXUtils.mapReduceTriplets(g, sendMsg, mergeMsg,
   │     Some((oldMessages, activeDirection))) [前回メッセージのactive情報を利用]
   ├─ messageCheckpointer.update(messages)
   ├─ isActiveMessagesNonEmpty = !messages.isEmpty()
   ├─ oldMessages.unpersist()
   └─ prevG.unpersistVertices() + prevG.edges.unpersist()

3. クリーンアップ
   ├─ messageCheckpointer.unpersistDataSet()
   ├─ graphCheckpointer.deleteAllCheckpoints()
   └─ messageCheckpointer.deleteAllCheckpoints()

4. 結果: g を返却
```

### フローチャート

```mermaid
flowchart TD
    A[入力グラフ + initialMsg] --> B[全頂点にinitialMsgでvprogを実行]
    B --> C[mapReduceTripletsでメッセージ生成・集約]
    C --> D{メッセージあり?}
    D -->|Yes| E{maxIterations未満?}
    E -->|Yes| F[joinVerticesでvprog実行]
    F --> G[mapReduceTripletsで新メッセージ]
    G --> H[旧メッセージ/グラフをunpersist]
    H --> D
    E -->|No| I[チェックポイントクリーンアップ]
    D -->|No| I
    I --> J[結果グラフ返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-79-01 | 初回全頂点実行 | 最初のスーパーステップでは全頂点がinitialMsgを受信しvprogを実行 | 初回のみ |
| BR-79-02 | アクティブ頂点のみ | 2回目以降はメッセージを受信した頂点のみvprogを実行 | 2回目以降 |
| BR-79-03 | activeDirection | 前回メッセージを受信した頂点に接続するエッジのみsendMsgを実行 | 2回目以降 |
| BR-79-04 | 可換結合法則 | mergeMsgは可換（a+b=b+a）かつ結合法則（(a+b)+c=a+(b+c)）を満たす必要あり | 全反復 |
| BR-79-05 | チェックポイント | spark.graphx.pregel.checkpointInterval設定によりリネージを定期的に切断 | 設定時 |

### 計算ロジック

- **初期化**: `graph.mapVertices((vid, vdata) => vprog(vid, vdata, initialMsg))` (Pregel.scala 131行目)
- **メッセージ生成**: `GraphXUtils.mapReduceTriplets(g, sendMsg, mergeMsg)` (137行目)
- **頂点更新**: `g.joinVertices(messages)(vprog)` (149行目)
- **アクティブメッセージ**: `GraphXUtils.mapReduceTriplets(g, sendMsg, mergeMsg, Some((oldMessages, activeDirection)))` (156-157行目)
- **チェックポイント間隔**: `spark.graphx.pregel.checkpointInterval` (129-130行目)

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

GraphXはデータベースを直接操作しない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalArgumentException | 入力検証 | maxIterations <= 0 | 正の整数を指定 |

### リトライ仕様

RDDの耐障害性メカニズムに従う。チェックポイント有効時はリネージが短縮されるため、障害復旧が効率的。

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

トランザクション機構は存在しない。

## パフォーマンス要件

- PeriodicGraphCheckpointerによりリネージの無制限な伸長を防止
- PeriodicRDDCheckpointerによりメッセージRDDのリネージも管理
- 各反復で前回のグラフとメッセージを明示的にunpersist
- activeDirection指定により不要なエッジのスキャンを回避
- activeFractionに基づくスキャン方式の動的選択（GraphImpl側で実装）

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

特になし。

## 備考

- GraphXのPregelは元のGoogleのPregelとは異なり、メッセージ送信をエッジ上で実行する
- joinVerticesは内部的にaggregateMessagesと同じ最適化を利用
- checkpointIntervalのデフォルトは-1（チェックポイント無効）

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Pregel.scala | `graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala` | 型パラメータ VD, ED, A（メッセージ型）の関係を理解する |

**読解のコツ**: Pregelの3つのユーザー定義関数（vprog, sendMsg, mergeMsg）の型シグネチャを最初に把握すること。vprogは頂点プログラム、sendMsgはエッジ上のメッセージ生成、mergeMsgはメッセージの結合。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | Pregel.scala | `graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala` | apply メソッド (116-177行目) - カリー化された引数パターン |

**主要処理フロー**:
- **116-124行目**: applyメソッドのシグネチャ。カリー化により設定パラメータ(graph, initialMsg, ...)と関数パラメータ(vprog, sendMsg, mergeMsg)を分離
- **126-127行目**: maxIterations > 0 のバリデーション
- **129-130行目**: checkpointInterval設定の取得
- **131行目**: 初期化 - 全頂点にinitialMsgでvprogを実行
- **132-134行目**: PeriodicGraphCheckpointer設定
- **137行目**: 最初のメッセージ生成
- **146-172行目**: メインループ - joinVertices + mapReduceTriplets + unpersist
- **162行目**: isEmpty()チェックによるアクティブメッセージ判定
- **173-176行目**: チェックポイントクリーンアップ

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

```
Pregel.apply(graph, initialMsg, maxIterations, activeDirection)(vprog, sendMsg, mergeMsg)
    |
    +-- graph.mapVertices(vprog(vid, vdata, initialMsg)) [初期化]
    +-- PeriodicGraphCheckpointer(checkpointInterval)
    +-- GraphXUtils.mapReduceTriplets(g, sendMsg, mergeMsg) [初回メッセージ]
    +-- PeriodicRDDCheckpointer(checkpointInterval)
    |
    +-- while (isActiveMessagesNonEmpty && i < maxIterations):
    |       +-- g.joinVertices(messages)(vprog) [頂点更新]
    |       +-- graphCheckpointer.update(g)
    |       +-- GraphXUtils.mapReduceTriplets(g, sendMsg, mergeMsg, active) [メッセージ生成]
    |       +-- messageCheckpointer.update(messages)
    |       +-- oldMessages.unpersist()
    |       +-- prevG.unpersistVertices() / edges.unpersist()
    |
    +-- messageCheckpointer.unpersistDataSet()
    +-- graphCheckpointer.deleteAllCheckpoints()
    +-- messageCheckpointer.deleteAllCheckpoints()
    +-- return g
```

### データフロー図

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

Graph[VD, ED]      初回: vprog(全頂点, initialMsg)
+ initialMsg              |
                   mapReduceTriplets(sendMsg)
                          |
                   mergeMsg(メッセージ集約)
                          |
                   joinVertices(vprog) [メッセージ受信頂点のみ]
                          |
                   mapReduceTriplets(sendMsg, active)
                          |
                        ... (反復)
                          |
                          v
                   Graph[VD, ED] (更新済み)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Pregel.scala | `graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala` | ソース | Pregelフレームワーク実装 |
| GraphXUtils.scala | `graphx/src/main/scala/org/apache/spark/graphx/GraphXUtils.scala` | ソース | mapReduceTriplets実装 |
| Graph.scala | `graphx/src/main/scala/org/apache/spark/graphx/Graph.scala` | ソース | joinVertices等のグラフ操作 |
| GraphImpl.scala | `graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala` | ソース | aggregateMessagesWithActiveSet実装 |
| PeriodicGraphCheckpointer.scala | `graphx/src/main/scala/org/apache/spark/graphx/util/PeriodicGraphCheckpointer.scala` | ソース | グラフチェックポイント管理 |
| PeriodicRDDCheckpointer.scala | `core/src/main/scala/org/apache/spark/rdd/util/PeriodicRDDCheckpointer.scala` | ソース | RDDチェックポイント管理 |
