# Apache Spark コードリーディングガイドライン

## はじめに

このガイドラインは、Apache Spark（バージョン 4.2.0-SNAPSHOT）のコードベースを効率的に理解するための手引きです。
Scala / Java / Python に精通していないエンジニアでも、段階的に学習できるよう構成されています。

**対象読者:**
- プロジェクトに新規参画するエンジニア
- 他言語からの経験者
- コードレビューを行う担当者

---

## 1. 言語基礎

> このセクションでは、Apache Spark の主要言語である Scala の基本構文と概念を解説します。Spark は Scala で記述されたコアエンジンに加え、Java API、Python API（PySpark）、R API も提供しています。

### 1.1 プログラム構造

Scala のプログラムはパッケージ宣言、インポート文、クラス/オブジェクト定義で構成されます。Spark ではパッケージ階層が `org.apache.spark` から始まります。

```scala
// ファイル: core/src/main/scala/org/apache/spark/SparkContext.scala:18-86
package org.apache.spark

import java.io._
import java.net.URI
import java.util.{Arrays, Locale, Properties, ServiceLoader, UUID}

import scala.collection.Map
import scala.reflect.{classTag, ClassTag}

class SparkContext(config: SparkConf) extends Logging {
  // クラス本体
}
```

**ポイント:**
- `package` でパッケージを宣言し、ディレクトリ構造と対応する
- `import` で外部クラスを取り込む（Java と同様だが `_` がワイルドカード）
- `class` でクラスを定義し、コンストラクタ引数を括弧内に記述する
- `extends` で継承、`with` でトレイトをミックスインする

### 1.2 データ型と変数

Scala では `val`（不変）と `var`（可変）で変数を宣言します。型推論が効くため、型を省略できる場合があります。

```scala
// ファイル: core/src/main/scala/org/apache/spark/SparkContext.scala:103-105
val startTime = System.currentTimeMillis()      // val: 不変（再代入不可）
private[spark] val stopped: AtomicBoolean = new AtomicBoolean(false)  // アクセス修飾子付き

// ファイル: core/src/main/scala/org/apache/spark/rdd/RDD.scala:152
@transient var name: String = _                  // var: 可変（再代入可能）、_ はデフォルト値
```

**主要なデータ型:**
- `Int`, `Long`, `Double`, `Boolean`, `String` -- 基本型
- `Option[T]` -- null の代わりに使用する（`Some(value)` / `None`）
- `Seq[T]`, `Array[T]`, `Map[K, V]` -- コレクション型
- `Future[T]` -- 非同期処理の結果

### 1.3 制御構造

Scala の制御構造は式として値を返します。

```scala
// パターンマッチ（Spark で頻出）
// ファイル: sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:118-130
logical match {
  case NameParameterizedQuery(compoundBody: CompoundBody, argNames, argValues) =>
    SqlScriptingExecution.executeSqlScript(
      session = sparkSession,
      script = compoundBody,
      args = argNames.zip(argValues).toMap)
  case compoundBody: CompoundBody =>
    SqlScriptingExecution.executeSqlScript(
      session = sparkSession,
      script = compoundBody)
  case _ => logical
}
```

**ポイント:**
- `match` / `case` はパターンマッチング。Java の `switch` の強化版
- `case class` の分解（デストラクチャリング）が可能
- `_` はワイルドカード（何にでもマッチ）

### 1.4 関数/メソッド定義

```scala
// ファイル: core/src/main/scala/org/apache/spark/rdd/RDD.scala:114-116
@DeveloperApi
def compute(split: Partition, context: TaskContext): Iterator[T]

// ファイル: core/src/main/scala/org/apache/spark/Partitioner.scala:42-44
abstract class Partitioner extends Serializable {
  def numPartitions: Int
  def getPartition(key: Any): Int
}
```

**ポイント:**
- `def` でメソッドを定義
- 戻り値の型は `: Type` で指定（省略可能だが、公開 API では明示が推奨）
- `abstract` メソッドは本体を持たない
- `lazy val` は初回アクセス時に遅延評価される（Spark で頻用）

### 1.5 モジュール/インポート

Spark ではパッケージ内アクセス修飾子が重要な役割を果たします。

```scala
// パッケージ内限定アクセス
private[spark] val stopped: AtomicBoolean  // spark パッケージ内からのみアクセス可能
private[sql] def planner                   // sql パッケージ内からのみアクセス可能

// 暗黙のインポート（implicit conversions）
import scala.language.implicitConversions   // 暗黙の変換を有効化
import scala.jdk.CollectionConverters._     // Java/Scala コレクション相互変換
```

**Scala 固有の概念:**
- `implicit` -- 暗黙のパラメータ/変換。型クラスパターンで多用
- `ClassTag[T]` -- 実行時の型情報を保持（JVM の型消去に対応）
- `@transient` -- シリアライズ対象外のフィールド（分散処理で重要）

---

## 2. プロジェクト固有の概念

> このセクションでは、Apache Spark 特有の概念を解説します。

### 2.1 フレームワーク固有の概念

#### RDD（Resilient Distributed Dataset）

Spark の基本抽象化。不変でパーティション分割されたコレクションを表し、並列処理が可能です。

```scala
// ファイル: core/src/main/scala/org/apache/spark/rdd/RDD.scala:56-82
/**
 * A Resilient Distributed Dataset (RDD), the basic abstraction in Spark.
 * Represents an immutable, partitioned collection of elements that can be
 * operated on in parallel.
 *
 * Internally, each RDD is characterized by five main properties:
 *  - A list of partitions
 *  - A function for computing each split
 *  - A list of dependencies on other RDDs
 *  - Optionally, a Partitioner for key-value RDDs
 *  - Optionally, a list of preferred locations to compute each split on
 */
abstract class RDD[T: ClassTag](
    @transient private var _sc: SparkContext,
    @transient private var deps: Seq[Dependency[_]]
  ) extends Serializable with Logging {
```

#### Dependency（依存関係）

RDD 間の依存関係を表し、NarrowDependency（パイプライン可能）と ShuffleDependency（シャッフル必要）に分かれます。

```scala
// ファイル: core/src/main/scala/org/apache/spark/Dependency.scala:40-61
@DeveloperApi
abstract class Dependency[T] extends Serializable {
  def rdd: RDD[T]
}

@DeveloperApi
abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] {
  def getParents(partitionId: Int): Seq[Int]
  override def rdd: RDD[T] = _rdd
}
```

#### SparkContext

Spark 機能のメインエントリーポイント。クラスタへの接続を表し、RDD、アキュムレータ、ブロードキャスト変数を生成します。

```scala
// ファイル: core/src/main/scala/org/apache/spark/SparkContext.scala:77-86
/**
 * Main entry point for Spark functionality. A SparkContext represents the
 * connection to a Spark cluster, and can be used to create RDDs, accumulators
 * and broadcast variables on that cluster.
 *
 * Only one SparkContext should be active per JVM.
 */
class SparkContext(config: SparkConf) extends Logging {
```

#### Catalyst オプティマイザ

Spark SQL のクエリ最適化フレームワーク。ルールベースの変換を論理プランに繰り返し適用します。

```scala
// ファイル: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:47-52
abstract class Optimizer(catalogManager: CatalogManager)
  extends RuleExecutor[LogicalPlan] with SQLConfHelper {
```

### 2.2 プロジェクト独自のパターン

#### Rule パターン

Catalyst エンジンの中核パターン。TreeNode に対する変換ルールを `Rule[TreeType]` として定義し、`RuleExecutor` がバッチ単位で適用します。

```scala
// ファイル: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/Rule.scala:24-36
abstract class Rule[TreeType <: TreeNode[_]] extends SQLConfHelper with Logging {
  protected lazy val ruleId = RuleIdCollection.getRuleId(this.ruleName)

  val ruleName: String = {
    val className = getClass.getName
    if (className endsWith "$") className.dropRight(1) else className
  }

  def apply(plan: TreeType): TreeType
}
```

#### LazyTry パターン

遅延評価と例外処理を組み合わせたパターン。QueryExecution で各フェーズの実行を遅延化するために使用されます。

```scala
// ファイル: sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:136-155
private val lazyAnalyzed = LazyTry {
  assertSqlScriptExecuted()
  try {
    val plan = executePhase(QueryPlanningTracker.ANALYSIS) {
      sparkSession.sessionState.analyzer.executeAndCheck(sqlScriptExecuted, tracker)
    }
    tracker.setAnalyzed(plan)
    plan
  } catch {
    case NonFatal(e) =>
      tracker.setAnalysisFailed(sqlScriptExecuted)
      throw e
  }
}

def analyzed: LogicalPlan = lazyAnalyzed.get
```

#### アノテーションによる API 分類

```scala
@DeveloperApi    // 開発者向けAPI（内部実装に近い）
@Experimental    // 実験的API（将来変更の可能性あり）
@Since("x.x")   // 導入バージョン
@transient       // シリアライズ対象外
```

---

## 3. 命名規則

> このセクションでは、プロジェクト全体で使用される命名規則を解説します。

### 3.1 ファイル・ディレクトリ命名

| パターン | 意味 | 例 |
|---------|------|-----|
| `XxxYyy.scala` | PascalCase でクラス名と一致 | `SparkContext.scala`, `DAGScheduler.scala` |
| `package.scala` | パッケージオブジェクト定義 | `core/.../rdd/package.scala` |
| `package-info.java` | Java 用パッケージ情報 | `core/.../rdd/package-info.java` |
| `Xxx.java` | Java ソース | `InternalRow.java` |
| `xxx_yyy.py` | Python のスネークケース | `java_gateway.py`, `rddsampler.py` |
| `src/main/scala/` | メインソースコード | 全モジュール共通 |
| `src/test/scala/` | テストコード | 全モジュール共通 |
| `src/main/resources/` | リソースファイル | 設定ファイル等 |

### 3.2 クラス・関数・変数命名

| プレフィックス/サフィックス | 意味 | 例 |
|---------------------------|------|-----|
| `Spark` + 名詞 | Spark のコア概念 | `SparkContext`, `SparkConf`, `SparkEnv` |
| 名詞 + `RDD` | RDD のサブクラス | `ShuffledRDD`, `MapPartitionsRDD`, `UnionRDD` |
| 名詞 + `Exec` | 物理実行プラン | `FilterExec`, `ProjectExec`, `SortExec` |
| 名詞 + `Manager` | リソース管理 | `BlockManager`, `ShuffleManager`, `ExecutorAllocationManager` |
| 名詞 + `Scheduler` | スケジューリング | `DAGScheduler`, `TaskScheduler` |
| 名詞 + `Backend` | バックエンド実装 | `SchedulerBackend`, `LocalSchedulerBackend` |
| 名詞 + `Listener` | イベントリスナー | `SparkListener`, `JobListener` |
| 名詞 + `Source` | メトリクスソース | `DAGSchedulerSource`, `BlockManagerSource` |
| 名詞 + `Utils` | ユーティリティ | `Utils`, `RPackageUtils`, `StandaloneResourceUtils` |
| `is` / `has` / `can` + 動詞 | Boolean 判定メソッド | `isSqlScript`, `isLazyAnalysis` |
| `get` + 名詞 | 取得メソッド | `getPartition`, `getPartitions`, `getPreferredLocations` |
| `assert` + 名詞/形容詞 | 検証メソッド（失敗時例外） | `assertAnalyzed`, `assertNotStopped` |
| `private[spark]` | Spark パッケージ内限定 | パッケージプライベート API |
| `private[sql]` | SQL パッケージ内限定 | SQL モジュール内部 API |

### 3.3 プログラム分類一覧

| 分類 | 配置場所 | 説明 |
|------|---------|------|
| コアエンジン | `core/` | RDD、スケジューラ、ストレージ等の基盤 |
| SQL エンジン（Catalyst） | `sql/catalyst/` | SQL パーサ、アナライザ、オプティマイザ |
| SQL エンジン（実行） | `sql/core/` | 物理プラン、実行エンジン |
| Spark Connect | `sql/connect/` | クライアント/サーバー分離アーキテクチャ |
| 機械学習 | `mllib/`, `mllib-local/` | MLlib 機械学習ライブラリ |
| グラフ処理 | `graphx/` | GraphX グラフ処理ライブラリ |
| ストリーミング | `streaming/` | Spark Streaming（DStream ベース） |
| コネクタ | `connector/` | 外部データソース連携（Kafka, Avro 等） |
| PySpark | `python/pyspark/` | Python API |
| 共通モジュール | `common/` | ネットワーク、シリアライゼーション等の共通基盤 |

---

## 4. ディレクトリ構造

> このセクションでは、プロジェクトのディレクトリ構造を解説します。

```
spark-master/
├── assembly/          # 配布用パッケージのアセンブリ
├── bin/               # 実行スクリプト（spark-shell, spark-submit 等）
├── build/             # ビルドスクリプト（Maven ラッパー等）
├── common/            # 共通モジュール群
│   ├── kvstore/       # キー・バリューストア
│   ├── network-common/# ネットワーク共通
│   ├── network-shuffle/# シャッフルネットワーク
│   ├── sketch/        # 確率的データ構造
│   ├── tags/          # テスト分類タグ
│   ├── unsafe/        # Off-heap メモリ操作
│   ├── utils/         # Scala ユーティリティ
│   ├── utils-java/    # Java ユーティリティ
│   └── variant/       # Variant データ型
├── conf/              # 設定テンプレート
├── connector/         # 外部データソースコネクタ
│   ├── avro/          # Apache Avro
│   ├── kafka-0-10/    # Apache Kafka
│   ├── kafka-0-10-sql/# Kafka SQL 統合
│   ├── protobuf/      # Protocol Buffers
│   └── ...
├── core/              # Spark コアエンジン
│   └── src/main/scala/org/apache/spark/
│       ├── api/       # Java/Python/R API
│       ├── broadcast/ # ブロードキャスト変数
│       ├── deploy/    # デプロイ・サブミット
│       ├── executor/  # エグゼキュータ
│       ├── rdd/       # RDD 実装
│       ├── rpc/       # RPC フレームワーク
│       ├── scheduler/ # ジョブ・タスクスケジューラ
│       ├── serializer/# シリアライゼーション
│       ├── shuffle/   # シャッフル
│       ├── storage/   # ブロックストレージ
│       └── util/      # ユーティリティ
├── data/              # テスト用サンプルデータ
├── dev/               # 開発者ツール
├── docs/              # ドキュメント
├── examples/          # サンプルアプリケーション
├── graphx/            # GraphX グラフ処理
├── launcher/          # アプリケーションランチャー
├── mllib/             # MLlib 機械学習
├── mllib-local/       # MLlib ローカル実装
├── project/           # SBT ビルド設定
├── python/            # PySpark
│   └── pyspark/
│       ├── sql/       # SQL API
│       ├── ml/        # 機械学習
│       ├── pandas/    # pandas API on Spark
│       └── streaming/ # ストリーミング
├── R/                 # SparkR
├── repl/              # Spark Shell（REPL）
├── resource-managers/ # リソースマネージャ統合
├── sbin/              # 管理スクリプト
├── sql/               # Spark SQL
│   ├── api/           # SQL API 定義
│   ├── catalyst/      # Catalyst オプティマイザ
│   ├── connect/       # Spark Connect
│   ├── core/          # SQL 実行エンジン
│   ├── hive/          # Hive 統合
│   ├── hive-thriftserver/ # Hive ThriftServer
│   └── pipelines/     # SQL パイプライン
├── streaming/         # Spark Streaming
├── tools/             # 開発ツール
└── ui-test/           # UI テスト
```

### 各ディレクトリの役割

| ディレクトリ | 役割 | 主要ファイル |
|-------------|------|-------------|
| `core/` | Spark のコアエンジン。RDD、スケジューリング、ストレージ、シャッフルの実装 | `SparkContext.scala`, `RDD.scala`, `DAGScheduler.scala` |
| `sql/catalyst/` | SQL クエリの解析・最適化フレームワーク（Catalyst エンジン） | `Optimizer.scala`, `Analyzer.scala`, `Rule.scala` |
| `sql/core/` | SQL クエリの物理実行計画と実行エンジン | `QueryExecution.scala`, `SparkSession.scala` |
| `sql/connect/` | Spark Connect のクライアント/サーバー通信 | gRPC ベースのサーバー/クライアント |
| `common/unsafe/` | Off-heap メモリ管理と低レベルデータ操作 | `UnsafeRow`, `Platform` |
| `common/network-common/` | ノード間ネットワーク通信の共通基盤 | RPC、データ転送 |
| `streaming/` | マイクロバッチベースのストリーム処理 | `StreamingContext.scala`, `DStreamGraph.scala` |
| `mllib/` | 分散機械学習アルゴリズム | `Pipeline.scala`, `Transformer.scala` |
| `connector/` | 外部データソースとの連携 | Kafka, Avro, Protobuf コネクタ |
| `python/pyspark/` | Python API。Py4J 経由で JVM と通信 | `__init__.py`, `sql/`, `ml/` |
| `examples/` | 各言語のサンプルアプリケーション | `SparkPi.scala`, `sql/` 配下の例 |

---

## 5. アーキテクチャ

> このセクションでは、プロジェクトのアーキテクチャパターンを解説します。

### 5.1 全体アーキテクチャ

Apache Spark はレイヤードアーキテクチャとモジュラーモノリスのハイブリッド構成を採用しています。コアエンジンの上に複数の高レベルライブラリ（SQL、MLlib、GraphX、Streaming）が積み重なっています。

```
┌─────────────────────────────────────────────────────────────────┐
│                    ユーザー API 層                                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────────┐ │
│  │ Spark SQL│  │  MLlib   │  │  GraphX  │  │ Structured       │ │
│  │ DataFrame│  │          │  │          │  │ Streaming        │ │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────────────┘ │
├───────┼──────────────┼──────────────┼──────────────┼─────────────┤
│       └──────────────┴──────┬───────┴──────────────┘             │
│                    ┌────────┴────────┐                           │
│                    │   Catalyst      │ SQL クエリ最適化           │
│                    │   Optimizer     │                           │
│                    └────────┬────────┘                           │
├─────────────────────────────┼───────────────────────────────────┤
│                    ┌────────┴────────┐                           │
│                    │    Spark Core   │ RDD, スケジューリング      │
│                    │   (DAGScheduler,│ ストレージ, シャッフル     │
│                    │    TaskScheduler│                           │
│                    │    BlockManager)│                           │
│                    └────────┬────────┘                           │
├─────────────────────────────┼───────────────────────────────────┤
│               ┌─────────────┴─────────────┐                     │
│               │    Cluster Manager        │                     │
│  ┌────────┐  ┌────────┐  ┌────────┐  ┌────────┐               │
│  │Standalone│ │  YARN  │  │  Mesos │  │Kubernetes│              │
│  └────────┘  └────────┘  └────────┘  └────────┘               │
└─────────────────────────────────────────────────────────────────┘
```

### 5.2 レイヤー構成

| レイヤー | 責務 | 代表的なファイル |
|---------|------|-----------------|
| ユーザー API | DataFrame/Dataset/RDD API の提供 | `sql/api/`, `core/.../rdd/RDD.scala` |
| SQL 最適化（Catalyst） | 論理プラン → 最適化 → 物理プラン変換 | `sql/catalyst/.../optimizer/Optimizer.scala` |
| 実行エンジン | 物理プランの実行、コード生成 | `sql/core/.../execution/QueryExecution.scala` |
| スケジューリング | DAG の Stage 分割とタスク配分 | `core/.../scheduler/DAGScheduler.scala` |
| ストレージ | ブロックの管理とキャッシュ | `core/.../storage/BlockManager.scala` |
| シャッフル | データの再分配 | `core/.../shuffle/ShuffleManager.scala` |
| ネットワーク | ノード間通信 | `common/network-common/` |
| クラスタ管理 | リソース確保とエグゼキュータ管理 | `resource-managers/` |

### 5.3 データフロー

#### RDD の実行フロー

```
ユーザーコード（RDD API 呼び出し）
    │
    ▼
SparkContext.runJob()
    │
    ▼
DAGScheduler: RDD の依存関係グラフを Stage に分割
    │  - ShuffleDependency の境界で Stage を切る
    │  - NarrowDependency はパイプライン化
    ▼
TaskScheduler: Stage 内の Task をエグゼキュータに配分
    │
    ▼
Executor: Task を実行し、RDD.compute() を呼び出す
    │
    ▼
結果を Driver に返却 / Shuffle ファイルに書き出し
```

#### SQL クエリの実行フロー（Catalyst パイプライン）

```
SQL文 / DataFrame API
    │
    ▼
Parser: SQL → 未解決論理プラン（Unresolved LogicalPlan）
    │
    ▼
Analyzer: カタログ参照でカラム名・テーブル名を解決
    │
    ▼
Optimizer: ルールベース最適化（述語プッシュダウン、定数畳み込み等）
    │
    ▼
Planner: 論理プラン → 物理プラン（SparkPlan）
    │
    ▼
Physical Plan Preparation: Exchange 挿入、バケッティング最適化等
    │
    ▼
Execution: コード生成 → RDD として実行
```

---

## 6. 主要コンポーネント

> このセクションでは、主要なコンポーネントとその連携を解説します。

### 6.1 エントリーポイント

#### SparkSubmit（アプリケーション起動）

```scala
// ファイル: core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala:48-54
/**
 * Main gateway of launching a Spark application.
 *
 * This program handles setting up the classpath with relevant Spark dependencies and provides
 * a layer over the different cluster managers and deploy modes that Spark supports.
 */
```

`bin/spark-submit` コマンドから呼び出され、アプリケーションの起動を担います。

#### SparkContext（コア API のエントリーポイント）

```scala
// ファイル: core/src/main/scala/org/apache/spark/SparkContext.scala:77-86
/**
 * Main entry point for Spark functionality. A SparkContext represents the connection to a Spark
 * cluster, and can be used to create RDDs, accumulators and broadcast variables on that cluster.
 */
class SparkContext(config: SparkConf) extends Logging {
```

JVM あたり1つだけアクティブにできます。RDD の生成、ジョブの投入を行います。

#### QueryExecution（SQL のエントリーポイント）

```scala
// ファイル: sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:58-72
/**
 * The primary workflow for executing relational queries using Spark. Designed to allow easy
 * access to the intermediate phases of query execution for developers.
 */
class QueryExecution(
    val sparkSession: SparkSession,
    val logical: LogicalPlan,
    val tracker: QueryPlanningTracker = new QueryPlanningTracker,
    ...
```

SQL クエリの解析、最適化、物理プラン生成、実行の全フェーズを管理します。

### 6.2 ビジネスロジック

#### DAGScheduler（ジョブスケジューリング）

```scala
// ファイル: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:58-79
/**
 * The high-level scheduling layer that implements stage-oriented scheduling. It computes a DAG of
 * stages for each job, keeps track of which RDDs and stage outputs are materialized, and finds a
 * minimal schedule to run the job.
 *
 * Spark stages are created by breaking the RDD graph at shuffle boundaries. RDD operations with
 * "narrow" dependencies, like map() and filter(), are pipelined together into one set of tasks
 * in each stage, but operations with shuffle dependencies require multiple stages.
 */
```

#### Catalyst Optimizer（クエリ最適化）

```scala
// ファイル: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:47-76
abstract class Optimizer(catalogManager: CatalogManager)
  extends RuleExecutor[LogicalPlan] with SQLConfHelper {

  protected def fixedPoint =
    FixedPoint(
      conf.optimizerMaxIterations,
      maxIterationsSetting = SQLConf.OPTIMIZER_MAX_ITERATIONS.key)
```

`RuleExecutor` を継承し、最適化ルールをバッチ単位で繰り返し適用します。FixedPoint（収束するまで）と Once（1回のみ）の実行戦略があります。

### 6.3 データアクセス

#### BlockManager（ブロックストレージ）

```
core/src/main/scala/org/apache/spark/storage/BlockManager.scala
```

RDD のパーティションデータ（ブロック）をメモリ・ディスクに保存し、リモートノードとの転送を管理します。

#### ShuffleManager（シャッフル管理）

```
core/src/main/scala/org/apache/spark/shuffle/ShuffleManager.scala
```

Stage 間のデータ再分配（シャッフル）を管理します。ソートベースのシャッフル実装がデフォルトです。

#### Connector（外部データソース）

```
connector/kafka-0-10-sql/ -- Kafka ストリーミングデータソース
connector/avro/           -- Avro ファイルフォーマット
connector/protobuf/       -- Protocol Buffers フォーマット
```

### 6.4 ユーティリティ/共通機能

#### common/unsafe

```
common/unsafe/src/main/java/org/apache/spark/unsafe/
```

Off-heap メモリ管理。`UnsafeRow` は Tungsten プロジェクトの一部で、行データのバイナリ表現を提供します。

#### common/network-common

ノード間の RPC 通信とデータ転送の基盤。Netty ベースで実装されています。

#### core/.../util/Utils.scala

```
core/src/main/scala/org/apache/spark/util/Utils.scala
```

ファイル操作、クラスローディング、文字列処理など、プロジェクト全体で使用される汎用ユーティリティです。

---

## 7. よく使われるパターン

> このセクションでは、コード内で頻出するパターンを解説します。

### パターン一覧

| パターン | 説明 | 出現頻度 | 代表的なファイル |
|---------|------|---------|-----------------|
| Rule パターン | TreeNode への変換ルール定義 | 高 | `sql/catalyst/.../rules/Rule.scala` |
| RuleExecutor バッチ | ルールをバッチ単位で実行 | 高 | `sql/catalyst/.../optimizer/Optimizer.scala` |
| Template Method | 抽象クラスでフレームワークを定義 | 高 | `core/.../rdd/RDD.scala` |
| Listener/Event | イベント駆動のフック機構 | 高 | `core/.../scheduler/SparkListener.scala` |
| Companion Object | クラスとオブジェクトのペア | 高 | `SparkContext`, `Partitioner` 等 |
| 型パラメータ + ClassTag | 型安全な分散コレクション | 高 | `RDD[T: ClassTag]` |
| LazyTry | 遅延評価 + 例外ハンドリング | 中 | `QueryExecution.scala` |
| private[xxx] アクセス | パッケージスコープ限定 | 高 | プロジェクト全体 |

### 各パターンの詳細

#### パターン1: Rule パターン（Catalyst の中核）

**目的:** SQL クエリプランを段階的に変換する。各最適化ルールを独立したクラスとして定義し、組み合わせ可能にする。

**実装例:**
```scala
// ファイル: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/Rule.scala:24-36
abstract class Rule[TreeType <: TreeNode[_]] extends SQLConfHelper with Logging {
  protected lazy val ruleId = RuleIdCollection.getRuleId(this.ruleName)
  val ruleName: String = { ... }
  def apply(plan: TreeType): TreeType
}
```

**解説:** 各オプティマイザルール（例: `PropagateEmptyRelation`, `PushDownLeftSemiAntiJoin`）は `Rule[LogicalPlan]` を継承し、`apply` メソッドでプランを変換します。`RuleExecutor` が複数のルールをバッチ単位で適用し、収束するまで繰り返します。

#### パターン2: Template Method（RDD サブクラス）

**目的:** RDD の基本フレームワークを定義し、サブクラスで計算ロジックをカスタマイズする。

**実装例:**
```scala
// ファイル: core/src/main/scala/org/apache/spark/rdd/RDD.scala:108-139
// サブクラスが実装すべきメソッド:
def compute(split: Partition, context: TaskContext): Iterator[T]
protected def getPartitions: Array[Partition]
protected def getDependencies: Seq[Dependency[_]] = deps
protected def getPreferredLocations(split: Partition): Seq[String] = Nil
```

**解説:** `RDD` 抽象クラスは5つの基本プロパティ（パーティション、計算関数、依存関係、パーティショナー、ロケーション優先度）を定義し、サブクラス（`MapPartitionsRDD`, `ShuffledRDD` 等）がこれらを実装します。

#### パターン3: Listener/Event パターン

**目的:** ジョブ実行の各ステージでイベントを発行し、UI やメトリクス収集などのフック処理を行う。

**実装例:**
```
// ファイル: core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
// SparkListenerInterface を実装して各イベントをハンドリング
// イベント例: onJobStart, onJobEnd, onStageCompleted, onTaskEnd
```

**解説:** `LiveListenerBus` がイベントバスとして機能し、登録されたリスナーにイベントを配信します。UI（`SparkUI`）、History Server、カスタムメトリクス収集に使用されます。

---

## 8. 業務フロー追跡の実践例

> このセクションでは、実際の業務フローをコードで追跡する方法を解説します。

### 8.1 フロー追跡の基本手順

1. エントリーポイントを特定（`SparkSubmit`, `SparkContext`, `SparkSession`）
2. 処理の流れを追跡（呼び出し関係を追う）
3. データの変換を確認（LogicalPlan -> PhysicalPlan -> RDD）
4. 最終的な出力を確認

### 8.2 フロー追跡の実例

#### 例1: SQL クエリ実行フロー

**概要:** `spark.sql("SELECT * FROM table WHERE id > 10")` が実行されてから結果が返るまでの処理フロー

**処理フロー:**
```
SparkSession.sql() → Parser → Analyzer → Optimizer → Planner → Execution
```

**詳細な追跡:**

1. **SQL のパース** (`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/`)
   ```scala
   // CatalystSqlParser / SparkSqlParser が SQL 文字列を未解決論理プランに変換
   // AbstractSqlParser.scala でパース処理のエントリーポイントを定義
   ```

2. **分析（Analyzer）** (`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala`)
   ```scala
   // テーブル名の解決、カラム名の解決、型チェック等を実行
   // RuleExecutor のバッチとしてルールを適用
   ```

3. **最適化（Optimizer）** (`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:47-76`)
   ```scala
   abstract class Optimizer(catalogManager: CatalogManager)
     extends RuleExecutor[LogicalPlan] with SQLConfHelper {
     // 述語プッシュダウン、カラムプルーニング、定数畳み込み等のルールを適用
   }
   ```

4. **物理プラン生成** (`sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:80`)
   ```scala
   protected def planner = sparkSession.sessionState.planner
   // 論理プランを物理プラン（SparkPlan）に変換
   ```

5. **実行** (`sql/core/src/main/scala/org/apache/spark/sql/execution/`)
   ```scala
   // SparkPlan が RDD に変換され、DAGScheduler / TaskScheduler を通じて実行
   // 結果は collect() で Driver に集約、または write() で外部に書き出し
   ```

#### 例2: RDD ジョブ投入フロー

**概要:** `rdd.count()` のようなアクション呼び出しからタスク実行完了までのフロー

**処理フロー:**
```
RDD.count() → SparkContext.runJob() → DAGScheduler.submitJob() → TaskScheduler → Executor
```

**詳細な追跡:**

1. **アクション呼び出し** (`core/src/main/scala/org/apache/spark/rdd/RDD.scala`)
   ```scala
   // count() は内部で SparkContext.runJob() を呼び出す
   ```

2. **DAG 分割** (`core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:58-79`)
   ```scala
   // RDD の依存関係グラフを解析
   // ShuffleDependency の境界で ResultStage / ShuffleMapStage に分割
   ```

3. **タスクスケジューリング** (`core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala`)
   ```scala
   // Stage 内の各パーティションに対して Task を生成
   // データロケーションを考慮してエグゼキュータに割り当て
   ```

4. **タスク実行** (`core/src/main/scala/org/apache/spark/executor/`)
   ```scala
   // Executor が Task を受信し、RDD.compute() を呼び出して結果を計算
   // 結果は BlockManager 経由で返却 / 保存
   ```

### 8.3 フロー追跡チェックリスト

- [ ] エントリーポイントを特定したか
- [ ] 呼び出し関係を把握したか
- [ ] データの変換ポイントを確認したか
- [ ] エラーハンドリングを確認したか
- [ ] 最終的な出力を確認したか

---

## 9. 設計書の参照順序

> このセクションでは、プロジェクト理解のための設計書参照順序を案内します。

### 9.1 目的別ロードマップ

#### 全体像を把握したい場合
1. `README.md` -- プロジェクト概要とビルド方法
2. 本ガイドライン（コードリーディングガイドライン）
3. `docs/code-to-docs/アーキテクチャ設計書/` -- アーキテクチャの詳細
4. `docs/code-to-docs/機能一覧/` -- 機能の全体像

#### 特定機能を理解したい場合
1. `docs/code-to-docs/機能一覧/` -- 機能の特定
2. `docs/code-to-docs/機能設計書/` -- 機能の詳細設計
3. 該当モジュールのソースコード（`core/`, `sql/`, `streaming/` 等）

#### 改修作業を行う場合
1. 対象機能の設計書
2. `docs/code-to-docs/API設計書/` -- API インターフェースの確認
3. `CONTRIBUTING.md` -- コントリビューションガイドライン
4. 該当モジュールのテストコード（`src/test/scala/`）

### 9.2 ドキュメント一覧

| ドキュメント | 概要 | 参照タイミング |
|-------------|------|---------------|
| `README.md` | プロジェクト概要、ビルド手順、クイックスタート | 最初に参照 |
| `CONTRIBUTING.md` | コントリビューション手順、コーディング規約 | 改修・貢献時 |
| `docs/code-to-docs/アーキテクチャ設計書/` | システム全体のアーキテクチャ | 全体像把握 |
| `docs/code-to-docs/機能設計書/` | 各機能の詳細設計 | 機能理解時 |
| `docs/code-to-docs/API設計書/` | API インターフェース仕様 | API 利用・改修時 |
| `docs/code-to-docs/データベース設計書/` | データモデル | データ関連の理解 |
| `docs/code-to-docs/セキュリティ設計書/` | セキュリティ設計 | セキュリティ確認時 |
| Apache Spark 公式ドキュメント | 公式 API ドキュメント | 常時参照 |

---

## 10. トラブルシューティング

> このセクションでは、コードリーディング時によくある問題と解決法を解説します。

### よくある疑問と回答

#### Q: implicit（暗黙の変換/パラメータ）がどこから来ているのか分からない
A: Scala の IDE（IntelliJ IDEA + Scala プラグイン）を使用すると、暗黙のパラメータの出所を表示できます。Spark では `import scala.jdk.CollectionConverters._` や `import scala.language.implicitConversions` が頻出します。また、`RDD` クラスでは `PairRDDFunctions` への暗黙の変換が `package.scala` で定義されています（`core/src/main/scala/org/apache/spark/rdd/package.scala` 参照）。

#### Q: private[spark] や private[sql] の意味は？
A: Scala のパッケージスコープアクセス修飾子です。`private[spark]` は `org.apache.spark` パッケージ内のすべてのクラスからアクセス可能であることを意味します。外部からは非公開 API として扱われます。公開 API は `@DeveloperApi` や `@Experimental` アノテーションで分類されています。

#### Q: @transient が付いたフィールドの意味は？
A: `@transient` はシリアライズ対象外のフィールドを示します。Spark では RDD やタスクがネットワーク越しにシリアライズされるため、ドライバー側でのみ必要なフィールド（例: `SparkContext` への参照）に付与されます。`RDD.scala` の `_sc` フィールドが典型例です。

#### Q: Catalyst の TreeNode 変換がどう動くのか分からない
A: Catalyst では `TreeNode.transform` / `TreeNode.transformDown` / `TreeNode.transformUp` メソッドがツリーの各ノードに対してパターンマッチングを適用します。`Rule.apply` メソッドの内部でこれらを使い、マッチしたノードを新しいノードに置き換えます。まずは `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/` を参照してください。

#### Q: テストの実行方法は？
A: Maven を使用します。全体テスト: `./build/mvn test`。特定モジュール: `./build/mvn test -pl core`。特定テスト: `./build/mvn test -pl core -Dtest=xxx`。SBT も利用可能です。

### デバッグのヒント

- **QueryExecution.explain()** を使用すると、SQL クエリの各フェーズ（Parsed, Analyzed, Optimized, Physical）のプランを表示できます
- **SparkUI**（デフォルトポート 4040）でジョブ、ステージ、タスクの状況をリアルタイムに確認できます
- **DAGScheduler** のログ（`log4j` の `org.apache.spark.scheduler` パッケージ）でステージ分割とタスク割り当てを追跡できます
- **spark.sql.planChangeLog.level=WARN** を設定すると、Catalyst オプティマイザのルール適用ログを確認できます

---

## 付録

### A. 用語集

| 用語 | 説明 |
|-----|------|
| RDD | Resilient Distributed Dataset。Spark の基本分散データ抽象化 |
| DataFrame | 名前付きカラムを持つ分散データセット。RDD の上に構築 |
| Dataset | 型安全な DataFrame（Scala/Java のみ） |
| Partition | RDD のデータ分割単位。各パーティションが1つのタスクで処理される |
| Stage | DAG のシャッフル境界で分割された実行単位 |
| Task | 1つのパーティションに対する計算単位 |
| Job | 1つのアクション（count, collect 等）に対応する計算全体 |
| Shuffle | Stage 間のデータ再分配操作 |
| Narrow Dependency | 親 RDD の各パーティションが子 RDD の高々1つのパーティションに使われる依存関係 |
| Wide Dependency | 親 RDD の各パーティションが子 RDD の複数パーティションに使われる依存関係（シャッフル発生） |
| Catalyst | Spark SQL のクエリ最適化フレームワーク |
| Tungsten | Spark のメモリ管理・コード生成最適化プロジェクト |
| SparkContext | Spark アプリケーションのエントリーポイント（RDD API） |
| SparkSession | Spark SQL のエントリーポイント（DataFrame/Dataset API） |
| Driver | ユーザープログラムを実行し、SparkContext を保持するプロセス |
| Executor | ワーカーノード上でタスクを実行するプロセス |
| Broadcast | ドライバーからエグゼキュータに効率的に配布される読み取り専用変数 |
| Accumulator | エグゼキュータからドライバーに集約されるカウンター変数 |
| Spark Connect | クライアント/サーバー分離アーキテクチャ（gRPC ベース） |

### B. ファイル一覧

| ファイル/ディレクトリ | 説明 | 主な内容 |
|---------------------|------|---------|
| `pom.xml` | Maven ビルド定義 | 依存関係、モジュール一覧 |
| `project/SparkBuild.scala` | SBT ビルド定義 | SBT 用ビルド設定 |
| `core/src/main/scala/org/apache/spark/SparkContext.scala` | コアエントリーポイント | RDD 生成、ジョブ投入 |
| `core/src/main/scala/org/apache/spark/SparkConf.scala` | 設定管理 | Spark 設定パラメータ |
| `core/src/main/scala/org/apache/spark/rdd/RDD.scala` | RDD 抽象クラス | 基本操作（map, filter, reduce 等） |
| `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | DAG スケジューラ | Stage 分割、ジョブスケジューリング |
| `core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala` | タスクスケジューラ | タスク割り当て |
| `core/src/main/scala/org/apache/spark/storage/BlockManager.scala` | ブロック管理 | メモリ/ディスクストレージ |
| `core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala` | アプリ起動 | spark-submit の実装 |
| `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/Rule.scala` | Catalyst ルール基底 | ルール定義の基底クラス |
| `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala` | クエリ最適化 | 最適化ルールのバッチ定義 |
| `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala` | クエリ分析 | 名前解決、型チェック |
| `sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala` | クエリ実行 | 解析 → 最適化 → 実行の全フロー |

### C. 参考資料

- [Apache Spark 公式ドキュメント](https://spark.apache.org/documentation.html)
- [Spark SQL、DataFrames、Datasets ガイド](https://spark.apache.org/docs/latest/sql-programming-guide.html)
- [RDD プログラミングガイド](https://spark.apache.org/docs/latest/rdd-programming-guide.html)
- [Spark 設定リファレンス](https://spark.apache.org/docs/latest/configuration.html)
- [Scala 公式ドキュメント](https://docs.scala-lang.org/)
- [Spark: Cluster Computing with Working Sets（原論文）](http://people.csail.mit.edu/matei/papers/2012/nsdi_spark.pdf)
- [Spark SQL: Relational Data Processing in Spark](https://dl.acm.org/doi/10.1145/2723372.2742797)
- [Useful Developer Tools](https://spark.apache.org/developer-tools.html)
