# 機能設計書 71-グラフ構造データ管理

## 概要

本ドキュメントは、Apache Spark GraphXにおけるグラフ構造データ管理機能の設計を記述する。頂点RDD（VertexRDD）とエッジRDD（EdgeRDD）の組み合わせによる分散グラフ構造の生成、変換、結合操作の仕様を定義する。

### 本機能の処理概要

GraphXのグラフ構造データ管理は、大規模グラフデータの分散処理を実現するための中核機能である。Graph抽象クラスとその実装であるGraphImplを中心に、頂点とエッジをRDDとして管理し、グラフの変換・結合・部分グラフ抽出などの操作を提供する。

**業務上の目的・背景**：ソーシャルネットワーク分析、Webリンク解析、通信ネットワーク分析、生物学的ネットワーク解析など、大規模なグラフ構造データを扱う業務において、分散環境でのグラフ処理基盤が必要である。GraphXは、Sparkの既存のRDDインフラストラクチャ上に構築されたグラフ処理フレームワークとして、これらの課題を解決する。

**機能の利用シーン**：PageRank、連結成分検出、三角形カウントなどのグラフアルゴリズムの実行基盤として利用される。また、ファイルからのグラフ読み込み、グラフのパーティショニング、頂点・エッジ属性の変換、サブグラフ抽出、グラフ結合といったETL的な前処理にも使用される。

**主要な処理内容**：
1. Graph.apply / Graph.fromEdges / Graph.fromEdgeTuples によるグラフ構築
2. mapVertices / mapEdges / mapTriplets による頂点・エッジ属性の変換
3. subgraph / mask によるサブグラフ抽出
4. partitionBy によるエッジのリパーティショニング
5. aggregateMessages によるメッセージ集約（近傍集約）
6. outerJoinVertices / joinVertices による頂点属性の結合更新
7. persist / cache / checkpoint によるグラフの永続化・チェックポイント
8. GraphLoader.edgeListFile によるファイルからのグラフ読み込み

**関連システム・外部連携**：HDFS等の分散ファイルシステムからのグラフデータ読み込み、SparkのRDD基盤との連携。GraphXのアルゴリズムライブラリ（PageRank、ConnectedComponents等）はすべてこのグラフ構造データ管理機能の上に構築されている。

**権限による制御**：特にロールや権限による制御は行われない。SparkContextのアクセス権限に従う。

## 関連画面

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

## 機能種別

データ構造管理 / 計算処理基盤

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| vertices | RDD[(VertexId, VD)] | No | 頂点RDD（頂点IDと属性のペア） | VertexIdはLong型 |
| edges | RDD[Edge[ED]] | Yes | エッジRDD（ソースID、宛先ID、属性） | srcId, dstIdはLong型 |
| defaultVertexAttr | VD | No | エッジに含まれるが頂点RDDにない頂点のデフォルト属性 | 型パラメータVDに準拠 |
| edgeStorageLevel | StorageLevel | No | エッジの保存レベル（デフォルト: MEMORY_ONLY） | StorageLevelの有効値 |
| vertexStorageLevel | StorageLevel | No | 頂点の保存レベル（デフォルト: MEMORY_ONLY） | StorageLevelの有効値 |
| partitionStrategy | PartitionStrategy | No | パーティション戦略 | EdgePartition2D / EdgePartition1D / RandomVertexCut / CanonicalRandomVertexCut |

### 入力データソース

- ファイル入力: GraphLoader.edgeListFile によるテキストファイル（タブ区切りのエッジリスト形式）
- プログラマティック入力: RDD[(VertexId, VertexId)]、RDD[Edge[ED]]、RDD[(VertexId, VD)] からの構築

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Graph[VD, ED] | Graph | 頂点属性VD、エッジ属性EDを持つ分散グラフ |
| vertices | VertexRDD[VD] | 頂点RDD |
| edges | EdgeRDD[ED] | エッジRDD |
| triplets | RDD[EdgeTriplet[VD, ED]] | エッジトリプレット（エッジ+隣接頂点属性） |

### 出力先

メモリ上のRDD（永続化時はメモリ/ディスク/HDFS）

## 処理フロー

### 処理シーケンス

```
1. グラフ構築
   └─ Graph.apply / fromEdges / fromEdgeTuples でGraphImplを生成
      ├─ EdgeRDD.fromEdges でエッジパーティションを構築
      ├─ VertexRDD でRoutingTableを構築し頂点パーティションを生成
      └─ GraphImpl(VertexRDD, ReplicatedVertexView) を生成

2. グラフ変換
   └─ mapVertices / mapEdges / mapTriplets で新しいGraphを生成
      ├─ 型が変わらない場合は差分レプリケーション（インクリメンタル更新）
      └─ 型が変わる場合は全頂点の再レプリケーション

3. メッセージ集約（aggregateMessages）
   └─ エッジパーティション上でsendMsgを実行
      ├─ activeSetに基づきスキャン方式を選択（IndexScan / EdgeScan）
      ├─ 各パーティションでローカルpre-aggregation
      └─ vertices.aggregateUsingIndex で最終集約

4. グラフ結合（outerJoinVertices）
   └─ 頂点RDDと外部RDDのleftJoinを実行
      ├─ 型が変わらない場合は差分レプリケーション
      └─ 型が変わる場合は全頂点の再レプリケーション
```

### フローチャート

```mermaid
flowchart TD
    A[入力データ: vertices RDD + edges RDD] --> B[EdgeRDD.fromEdges]
    B --> C[EdgePartitionBuilder でエッジパーティション構築]
    C --> D[VertexRDD 構築: RoutingTable + VertexPartition]
    D --> E[GraphImpl 生成]
    E --> F{操作種別}
    F -->|変換| G[mapVertices / mapEdges / mapTriplets]
    F -->|フィルタ| H[subgraph / mask]
    F -->|集約| I[aggregateMessages]
    F -->|結合| J[outerJoinVertices]
    G --> K[新しい Graph を返却]
    H --> K
    I --> L[VertexRDD を返却]
    J --> K
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-71-01 | 頂点ID一意性 | 頂点IDはグラフ内で一意であること | グラフ構築時 |
| BR-71-02 | 不変性 | グラフ操作は新しいGraphオブジェクトを返す（元のグラフは変更されない） | 全変換操作 |
| BR-71-03 | 差分レプリケーション | 型が変わらないmapVerticesでは変更された頂点のみを再レプリケーション | mapVertices / outerJoinVertices |
| BR-71-04 | パーティション戦略 | groupEdgesを正しく動作させるにはpartitionByで事前パーティショニングが必要 | groupEdges使用時 |

### 計算ロジック

- **EdgePartition2D**: src, dstのハッシュに基づく2次元分割。頂点レプリケーションの上限は `2 * sqrt(numParts)`
- **EdgePartition1D**: srcのハッシュのみに基づく1次元分割。同じソースのエッジを同一パーティションに配置
- **aggregateMessages**: activeFractionが0.8未満の場合はIndexScanを使用し、それ以外はEdgeScanを使用（GraphImpl 210-238行目）

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

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

GraphXはデータベースを直接操作しない。すべてのデータはRDDとしてメモリ/ディスク上で管理される。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalArgumentException | 入力検証エラー | edgeListFileでフィールドが2未満の行 | 入力ファイルのフォーマットを修正 |
| SparkException | 実行時エラー | EdgeDirection.Bothを不正に使用 | EdgeDirection.Eitherを使用 |

### リトライ仕様

RDDの耐障害性メカニズムに従い、パーティション障害時はリネージに基づいて自動再計算される。

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

GraphXにはトランザクション機構は存在しない。グラフ操作は関数型で不変（イミュータブル）であり、各操作は新しいグラフを返す。

## パフォーマンス要件

- パーティション戦略の選択によりデータ局所性と頂点レプリケーションのトレードオフを制御
- aggregateMessagesでは、activeFractionに基づきIndexScanとEdgeScanを動的に切り替え（閾値0.8）
- チェックポイント機能によりリネージの切断が可能

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

SparkContextの認証・認可メカニズムに従う。グラフデータ自体に対する個別のアクセス制御機構は提供されていない。

## 備考

- GraphXはScala APIのみ提供（Python/Java向けの直接APIはない）
- VertexIdはLong型のエイリアス（64bit整数）
- GraphOps（暗黙変換）を通じてPageRank, ConnectedComponents等の便利メソッドが利用可能

---

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

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

### 推奨読解順序

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

まず、グラフの基本データ構造（頂点、エッジ、エッジトリプレット）を理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Edge.scala | `graphx/src/main/scala/org/apache/spark/graphx/Edge.scala` | エッジの基本構造（srcId, dstId, attr）を理解する |
| 1-2 | EdgeTriplet.scala | `graphx/src/main/scala/org/apache/spark/graphx/EdgeTriplet.scala` | Edgeを拡張し隣接頂点属性（srcAttr, dstAttr）を追加した構造を理解する |
| 1-3 | PartitionStrategy.scala | `graphx/src/main/scala/org/apache/spark/graphx/PartitionStrategy.scala` | エッジパーティショニング戦略（EdgePartition2D等）の分割ロジックを理解する |

**読解のコツ**: Edge.scalaの`@specialized`アノテーション（32行目）はプリミティブ型のボクシング回避のためのScala特有の最適化。EdgeTripletはEdgeを継承しており、エッジ情報に加えて隣接頂点の属性を保持する。

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

Graph抽象クラスとそのコンパニオンオブジェクトがグラフ操作の入口。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | Graph.scala | `graphx/src/main/scala/org/apache/spark/graphx/Graph.scala` | Graph抽象クラスの定義。vertices, edges, tripletsの3つのRDDプロパティと各種変換メソッドの契約を理解する |
| 2-2 | GraphOps.scala | `graphx/src/main/scala/org/apache/spark/graphx/GraphOps.scala` | 暗黙変換で提供される便利メソッド群。degrees計算、pregel、各種グラフアルゴリズムへの委譲 |

**主要処理フロー**:
1. **41行目**: Graph抽象クラスの定義。VD（頂点属性型）とED（エッジ属性型）の型パラメータ
2. **49行目**: vertices: VertexRDD[VD] - 頂点RDD
3. **62行目**: edges: EdgeRDD[ED] - エッジRDD
4. **80行目**: triplets: RDD[EdgeTriplet[VD, ED]] - エッジトリプレットRDD
5. **377-383行目**: aggregateMessages - 近傍メッセージ集約の主要API
6. **460行目**: Graph.apply - コンパニオンオブジェクトのファクトリメソッド
7. **541-542行目**: 暗黙変換 graphToGraphOps

#### Step 3: 実装クラスを理解する

GraphImplがGraph抽象クラスの実体実装。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | GraphImpl.scala | `graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala` | GraphのRDD実装。VertexRDDとReplicatedVertexViewで構成される |

**主要処理フロー**:
- **34-37行目**: GraphImplクラスの定義。VertexRDDとReplicatedVertexViewを保持
- **45-50行目**: tripletsの遅延評価実装
- **96-116行目**: partitionByの実装。HashPartitionerを使用してエッジを再分割
- **122-138行目**: mapVerticesの実装。型保存時は差分レプリケーション、非保存時は全再レプリケーション
- **188-244行目**: aggregateMessagesWithActiveSetの実装。activeFractionに基づくスキャン方式選択
- **270-356行目**: GraphImplコンパニオンオブジェクト。各種ファクトリメソッド

#### Step 4: ファイル入力を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | GraphLoader.scala | `graphx/src/main/scala/org/apache/spark/graphx/GraphLoader.scala` | edgeListFileメソッドによるテキストファイルからのグラフ読み込み |

**主要処理フロー**:
- **60-105行目**: edgeListFile。テキストファイルを読み込み、EdgePartitionBuilderでパーティションを構築してGraphImplを生成

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

```
Graph.apply / Graph.fromEdges
    |
    +-- GraphImpl.apply(vertices, edges, defaultVertexAttr, ...)
    |       |
    |       +-- EdgeRDD.fromEdges(edges)
    |       +-- VertexRDD(vertices, edgeRDD, defaultVertexAttr)
    |       +-- GraphImpl(vertexRDD, edgeRDD)
    |               +-- ReplicatedVertexView(EdgeRDDImpl)
    |
    +-- graph.mapVertices(f)
    |       +-- vertices.mapVertexPartitions
    |       +-- vertices.diff(newVerts) [型保存時]
    |       +-- replicatedVertexView.updateVertices [型保存時]
    |
    +-- graph.aggregateMessages(sendMsg, mergeMsg)
            +-- replicatedVertexView.upgrade
            +-- edges.partitionsRDD.mapPartitions
            |       +-- edgePartition.aggregateMessagesIndexScan / EdgeScan
            +-- vertices.aggregateUsingIndex(preAgg, mergeMsg)
```

### データフロー図

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

テキストファイル -----> GraphLoader.edgeListFile ------> Graph[Int, Int]
                           |
RDD[Edge[ED]] -------> Graph.fromEdges ------------> Graph[VD, ED]
                           |
RDD[(VertexId,VD)] --> Graph.apply -----------------> Graph[VD, ED]
+ RDD[Edge[ED]]           |
                           v
                    GraphImpl(VertexRDD, ReplicatedVertexView)
                           |
                    +------+------+
                    |             |
             mapVertices    aggregateMessages
                    |             |
                    v             v
             Graph[VD2, ED]  VertexRDD[A]
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Graph.scala | `graphx/src/main/scala/org/apache/spark/graphx/Graph.scala` | ソース | グラフ抽象クラス定義 |
| GraphOps.scala | `graphx/src/main/scala/org/apache/spark/graphx/GraphOps.scala` | ソース | 暗黙変換で提供される便利メソッド群 |
| GraphImpl.scala | `graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala` | ソース | Graphの実装クラス |
| Edge.scala | `graphx/src/main/scala/org/apache/spark/graphx/Edge.scala` | ソース | エッジデータ構造 |
| EdgeTriplet.scala | `graphx/src/main/scala/org/apache/spark/graphx/EdgeTriplet.scala` | ソース | エッジトリプレットデータ構造 |
| EdgeContext.scala | `graphx/src/main/scala/org/apache/spark/graphx/EdgeContext.scala` | ソース | aggregateMessagesのコンテキスト |
| EdgeDirection.scala | `graphx/src/main/scala/org/apache/spark/graphx/EdgeDirection.scala` | ソース | エッジ方向の列挙 |
| VertexRDD.scala | `graphx/src/main/scala/org/apache/spark/graphx/VertexRDD.scala` | ソース | 頂点RDD抽象クラス |
| EdgeRDD.scala | `graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala` | ソース | エッジRDD抽象クラス |
| PartitionStrategy.scala | `graphx/src/main/scala/org/apache/spark/graphx/PartitionStrategy.scala` | ソース | エッジパーティション戦略 |
| GraphLoader.scala | `graphx/src/main/scala/org/apache/spark/graphx/GraphLoader.scala` | ソース | ファイルからのグラフ読み込み |
| GraphXUtils.scala | `graphx/src/main/scala/org/apache/spark/graphx/GraphXUtils.scala` | ソース | ユーティリティ関数 |
| ReplicatedVertexView.scala | `graphx/src/main/scala/org/apache/spark/graphx/impl/ReplicatedVertexView.scala` | ソース | 頂点レプリケーション管理 |
| EdgePartition.scala | `graphx/src/main/scala/org/apache/spark/graphx/impl/EdgePartition.scala` | ソース | エッジパーティション実装 |
| EdgePartitionBuilder.scala | `graphx/src/main/scala/org/apache/spark/graphx/impl/EdgePartitionBuilder.scala` | ソース | エッジパーティション構築 |
| VertexPartition.scala | `graphx/src/main/scala/org/apache/spark/graphx/impl/VertexPartition.scala` | ソース | 頂点パーティション実装 |
| RoutingTablePartition.scala | `graphx/src/main/scala/org/apache/spark/graphx/impl/RoutingTablePartition.scala` | ソース | ルーティングテーブル |
