# 機能設計書 11-イベントログ記録

## 概要

本ドキュメントは、Apache Sparkのイベントログ記録機能について、設計仕様・処理フロー・コードリーディングガイドを記述した機能設計書である。

### 本機能の処理概要

Sparkアプリケーションの実行中に発生する各種イベント（ジョブの開始・完了、ステージの送信・完了、タスクの開始・終了、Executorの追加・削除など）をJSON形式でファイルに記録し、事後分析やHistory Serverによる可視化を可能にする。

**業務上の目的・背景**：分散処理システムでは、実行中のアプリケーションの挙動を事後的に分析する必要がある。特にパフォーマンスチューニング、障害分析、リソース利用状況の把握において、実行イベントの詳細な記録が不可欠である。イベントログ記録機能は、Sparkアプリケーションの全ライフサイクルにわたるイベントを永続化し、History Serverや外部分析ツールでの活用を実現する。

**機能の利用シーン**：(1) アプリケーション完了後にHistory Serverで実行履歴を閲覧する場合、(2) パフォーマンス問題の事後分析、(3) リソース利用パターンの分析、(4) 障害原因の調査時にイベントログを参照する場合。

**主要な処理内容**：
1. SparkListenerインターフェースを実装し、各種イベントをフックしてJSON形式にシリアライズ
2. EventLogFileWriterを介してログディレクトリにイベントを書き出し
3. ステージ完了時にExecutorメトリクスのピーク値を記録
4. 機密情報（パスワード等のSparkプロパティ）のリダクション処理
5. ブロック更新やExecutorメトリクスの記録を設定で制御

**関連システム・外部連携**：History Server（イベントログの読み取り先）、Hadoop FileSystem（HDFS等へのログ書き出し）

**権限による制御**：イベントログの書き込み先ディレクトリに対するファイルシステム権限に依存する。ログ内容の機密情報はリダクション処理される。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 13 | Driver Log（ドライバログ） | 補助機能 | ドライバログディレクトリからRollingFileAppenderのログファイルを取得・表示 |

## 機能種別

データ連携 / ログ出力

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| spark.eventLog.enabled | Boolean | No | イベントログ記録の有効化 | デフォルト: false |
| spark.eventLog.dir | String | No | イベントログ出力ディレクトリ | デフォルト: /tmp/spark-events |
| spark.eventLog.logBlockUpdates.enabled | Boolean | No | ブロック更新イベントの記録有無 | デフォルト: false |
| spark.eventLog.logStageExecutorMetrics | Boolean | No | ステージExecutorメトリクスの記録有無 | デフォルト: false |
| appId | String | Yes | アプリケーションID | 空でないこと |
| appAttemptId | Option[String] | No | アプリケーション試行ID | - |

### 入力データソース

SparkListenerバスから配信される各種SparkListenerEventオブジェクト（SparkListenerJobStart, SparkListenerStageCompleted, SparkListenerTaskEnd等）。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| eventJson | String | SparkListenerEventをJSON形式にシリアライズした文字列 |
| logFile | File | イベントログファイル（JSONLines形式） |

### 出力先

設定されたログディレクトリ（spark.eventLog.dir）配下のアプリケーション固有のサブディレクトリにJSONファイルとして出力。Hadoop FileSystem API経由で書き込み。

## 処理フロー

### 処理シーケンス

```
1. EventLoggingListenerの初期化
   └─ appId, appAttemptId, logBaseDir, sparkConfを受け取りインスタンス生成
2. start()の呼び出し
   └─ EventLogFileWriter.start()でログファイルを作成し、initEventLog()でSparkバージョン情報を書き込み
3. イベント受信と記録
   └─ SparkListenerBusからイベントを受信し、logEvent()でJSON化して書き出し
4. ステージ完了時のメトリクス記録
   └─ shouldLogStageExecutorMetricsがtrueの場合、ピーク値をSparkListenerStageExecutorMetricsとして記録
5. 機密情報のリダクション
   └─ 環境変数やジョブプロパティ内の機密情報をマスク
6. stop()の呼び出し
   └─ EventLogFileWriter.stop()でログファイルをクローズ
```

### フローチャート

```mermaid
flowchart TD
    A[SparkListener Event発生] --> B{イベント種別判定}
    B -->|Stage Submitted| C[logEvent + メトリクス初期化]
    B -->|Task End| D[logEvent + ピーク値更新]
    B -->|Stage Completed| E[メトリクスログ出力 + logEvent with flush]
    B -->|Job Start/End| F[logEvent with flush + プロパティリダクション]
    B -->|Environment Update| G[リダクション処理 + logEvent]
    B -->|Block Updated| H{shouldLogBlockUpdates?}
    H -->|Yes| I[logEvent with flush]
    H -->|No| J[スキップ]
    B -->|その他| K[logEvent with flush]
    C --> L[EventLogFileWriter.writeEvent]
    D --> L
    E --> L
    F --> L
    G --> L
    I --> L
    K --> L
    L --> M[JSONファイルに書き込み]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-11-01 | イベント記録制御 | spark.eventLog.enabledがtrueの場合のみイベントログを記録する | アプリケーション起動時 |
| BR-11-02 | フラッシュ制御 | ジョブ・ステージ単位の重要イベントはフラッシュ付きで書き込む | Stage Completed, Job Start/End等 |
| BR-11-03 | メトリクス記録制御 | spark.eventLog.logStageExecutorMetricsがtrueの場合のみExecutorメトリクスを記録 | ステージ完了時 |
| BR-11-04 | ブロック更新記録制御 | spark.eventLog.logBlockUpdates.enabledがtrueの場合のみブロック更新を記録 | ブロック更新イベント発生時 |
| BR-11-05 | 機密情報リダクション | Sparkプロパティ内の機密情報をマスクして記録する（Classpath Entriesは除外） | 環境情報・ジョブプロパティ記録時 |
| BR-11-06 | メトリクスピーク値追跡 | ステージ単位でExecutorごとのメトリクスピーク値を追跡し、ステージ完了時に記録 | shouldLogStageExecutorMetrics有効時 |

### 計算ロジック

メトリクスピーク値はExecutorMetrics.compareAndUpdatePeakValues()により、各メトリクス項目の最大値を保持する方式で追跡される。

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

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

本機能はデータベースを直接操作しない。ファイルシステムへの書き込みのみ行う。

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし |

### テーブル別操作詳細

該当なし（ファイルシステムベースの永続化）

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | IOException | ログファイルの書き込み失敗 | EventLogFileWriterが例外をスロー。アプリケーション自体は継続可能 |
| - | SerializationException | イベントのJSON変換失敗 | JsonProtocol.sparkEventToJsonStringが例外をスロー |

### リトライ仕様

リトライは行わない。書き込み失敗時はログ出力のみ。

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

ファイルベースの書き込みのため、トランザクション管理は行わない。イベントは順序通りに書き込まれ、フラッシュにより確実にディスクに反映される。

## パフォーマンス要件

イベント処理はSparkListenerBusの別スレッドで非同期に行われるため、メインの処理パスに直接的な影響を与えない。ただし、大量のイベントが発生する場合はI/O負荷に注意が必要。

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

- 機密情報（パスワード、トークン等）はUtils.redact()を用いてリダクション処理される
- Classpath Entriesはリダクション対象外
- ログファイルのアクセス権はファイルシステムの権限設定に依存

## 備考

- EventLogFileWriterは単一ファイル書き込みとローリングファイル書き込みの2種類が存在する
- テストモード(spark.eventLog.testing)ではメモリ内にもイベントを保持する
- ドライバーのExecutorメトリクス更新はダミーステージキー(-1, -1)で識別される

---

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

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

### 推奨読解順序

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

まず、SparkListenerイベントの型階層とJSON変換プロトコルを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | SparkListenerEvent各種イベントクラスの定義を確認 |
| 1-2 | ExecutorMetrics.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorMetrics.scala` | Executorメトリクスのデータ構造とcompareAndUpdatePeakValuesメソッドを理解 |

**読解のコツ**: SparkListenerはScalaのトレイトを使った型階層で、各イベントはcase classとして定義されている。`sealed trait`ではないため拡張可能。

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

EventLoggingListenerの初期化と起動フローを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | メインクラス。コンストラクタ引数、start()、各onXxxメソッドの実装を確認 |

**主要処理フロー**:
1. **48-54行目**: コンストラクタでappId, logBaseDir等を受け取り初期化
2. **63-64行目**: EventLogFileWriterの生成
3. **82-85行目**: start()でログファイル作成とメタデータ書き込み
4. **97-103行目**: logEvent()でイベントをJSON化してファイルに書き出し
5. **106-113行目**: onStageSubmitted()でメトリクス追跡の初期化
6. **136-158行目**: onStageCompleted()でピークメトリクスの書き出しとイベント記録

#### Step 3: JSON変換層を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | JsonProtocol.scala | `core/src/main/scala/org/apache/spark/util/JsonProtocol.scala` | sparkEventToJsonStringメソッドによるイベントのJSON変換処理 |

#### Step 4: ファイル書き込み層を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | EventLogFileWriters.scala | `core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala` | EventLogFileWriterのstart(), writeEvent(), stop()の実装 |

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

```
SparkContext (イベントログ有効化時)
    │
    ├─ EventLoggingListener (SparkListenerとして登録)
    │      ├─ start()
    │      │    └─ EventLogFileWriter.start() → ログファイル作成
    │      │    └─ initEventLog() → SparkListenerLogStartを書き込み
    │      │
    │      ├─ onXxxEvent() (各種イベントハンドラ)
    │      │    └─ logEvent()
    │      │         └─ JsonProtocol.sparkEventToJsonString()
    │      │         └─ EventLogFileWriter.writeEvent()
    │      │
    │      ├─ onStageCompleted()
    │      │    └─ SparkListenerStageExecutorMetrics記録
    │      │    └─ logEvent() with flush
    │      │
    │      └─ stop()
    │           └─ EventLogFileWriter.stop() → ログファイルクローズ
    │
    └─ SparkListenerBus (イベント配信)
```

### データフロー図

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

SparkListenerEvent ───▶ EventLoggingListener          ───▶ イベントログファイル
(各種イベント)              ├─ redactProperties()              (JSON Lines形式)
                           ├─ JsonProtocol変換
                           └─ EventLogFileWriter

ExecutorMetrics    ───▶ liveStageExecutorMetrics      ───▶ SparkListenerStageExecutorMetrics
(タスク終了時)             (ピーク値追跡HashMap)                (ステージ完了時に出力)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | ソース | イベントログ記録のメインクラス |
| EventLogFileWriters.scala | `core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala` | ソース | ログファイルの書き込み処理 |
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベント型定義 |
| JsonProtocol.scala | `core/src/main/scala/org/apache/spark/util/JsonProtocol.scala` | ソース | イベントのJSON変換 |
| ExecutorMetrics.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorMetrics.scala` | ソース | Executorメトリクスデータ構造 |
| config/package.scala | `core/src/main/scala/org/apache/spark/internal/config/package.scala` | 設定 | EVENT_LOG_*の設定定数定義 |
