# 帳票設計書 11-EventLoggingListener イベントログ

## 概要

本ドキュメントは、Apache Spark の `EventLoggingListener` によるイベントログ出力機能の帳票設計書である。Spark アプリケーション実行中に発生する各種イベント（ジョブ・ステージ・タスクの開始/終了、Executor の追加/削除等）を JSON 形式で永続ストレージに記録し、後から History Server で閲覧可能にする仕組みを定義する。

### 本帳票の処理概要

**業務上の目的・背景**：Spark アプリケーションの実行履歴を永続化し、アプリケーション終了後でも実行状況を振り返ることができるようにする。運用監視・障害調査・パフォーマンスチューニングにおいて、過去のジョブ実行状況を確認する必要があり、本帳票がその基盤データを提供する。History Server はこのイベントログを読み込んで Web UI を再構成する。

**帳票の利用シーン**：本帳票は Spark アプリケーション実行中にリアルタイムで出力される。アプリケーション完了後、管理者や開発者が History Server を通じてイベントログを閲覧し、ジョブの成功/失敗状況、ステージごとの処理時間、タスクの分布状況などを分析する場面で利用される。障害発生時のポストモーテム分析にも活用される。

**主要な出力内容**：
1. アプリケーションの開始・終了イベント（SparkListenerApplicationStart/End）
2. ジョブの開始・終了イベント（SparkListenerJobStart/End）
3. ステージの送信・完了イベント（SparkListenerStageSubmitted/Completed）
4. タスクの開始・終了イベント（SparkListenerTaskStart/End）
5. Executor の追加・削除・除外イベント
6. ブロック更新イベント（オプション）
7. ステージ別 Executor メトリクスのピーク値（オプション）
8. 環境情報更新イベント（機密情報はリダクト処理済み）

**帳票の出力タイミング**：`spark.eventLog.enabled=true` が設定された状態で Spark アプリケーションが起動すると、自動的にイベントログの記録が開始される。各イベント発生時にリアルタイムでログファイルに書き込まれる。ステージ完了やジョブ開始/終了などの重要イベントではフラッシュが実行される。

**帳票の利用者**：Spark クラスタ管理者、データエンジニア、アプリケーション開発者、SRE チーム

## 帳票種別

イベントログ（JSON形式ログファイル）

## 利用画面

| 画面No | 画面名 | URL/ルーティング | 出力操作 |
|--------|--------|-----------------|---------|
| - | History Server | http://{host}:18080 | アプリケーション一覧から選択して閲覧 |

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | JSON（1行1イベント形式、改行区切り） |
| 用紙サイズ | 該当なし（ファイル出力） |
| 向き | 該当なし |
| ファイル名 | `{appId}_{appAttemptId}` 形式（EventLogFileWriter による命名） |
| 出力方法 | 永続ストレージ（HDFS/S3/ローカルFS等）への書き込み |
| 文字コード | UTF-8 |

### PDF固有設定

該当なし（JSONファイル出力のため）

### Excel固有設定

該当なし（JSONファイル出力のため）

## 帳票レイアウト

### レイアウト概要

イベントログは JSON Lines 形式で出力される。各行が1つの Spark イベントを JSON オブジェクトとして表現する。

```
┌─────────────────────────────────────┐
│  ログ開始メタデータ (LogStart)        │
├─────────────────────────────────────┤
│  アプリケーション開始イベント          │
│  環境情報更新イベント                 │
│  Executor 追加イベント               │
│  ジョブ開始イベント                   │
│  ステージ送信イベント                 │
│  タスク開始/終了イベント（繰り返し）    │
│  ステージ完了イベント                 │
│  ジョブ終了イベント                   │
│  ...（イベント発生順に記録）           │
│  アプリケーション終了イベント          │
└─────────────────────────────────────┘
```

### ヘッダー部

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | Event | イベント種別名 | SparkListenerEvent クラス名 | 文字列 |
| 2 | Spark Version | Spark バージョン | SPARK_VERSION 定数 | 文字列（LogStart のみ） |

### 明細部

| No | 項目名 | 説明 | データ取得元 | 表示形式 | 列幅 |
|----|-------|------|-------------|---------|-----|
| 1 | Event | イベント種別 | SparkListenerEvent のクラス名 | 文字列 | - |
| 2 | Job ID | ジョブ識別子 | SparkListenerJobStart/End.jobId | 整数 | - |
| 3 | Stage Info | ステージ情報 | SparkListenerStageSubmitted/Completed.stageInfo | JSONオブジェクト | - |
| 4 | Task Info | タスク情報 | SparkListenerTaskEnd.taskInfo | JSONオブジェクト | - |
| 5 | Task Metrics | タスクメトリクス | SparkListenerTaskEnd.taskMetrics | JSONオブジェクト | - |
| 6 | Executor ID | Executor 識別子 | SparkListenerExecutorAdded/Removed.executorId | 文字列 | - |
| 7 | Block Update Info | ブロック更新情報 | SparkListenerBlockUpdated.blockUpdatedInfo | JSONオブジェクト | - |
| 8 | Properties | ジョブ/ステージプロパティ | リダクト処理済み Properties | JSONオブジェクト | - |

### フッター部

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | Application End | アプリケーション終了イベント | SparkListenerApplicationEnd | JSON |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| spark.eventLog.enabled | true に設定されていること | Yes |
| spark.eventLog.dir | イベントログ出力ディレクトリ（デフォルト: /tmp/spark-events） | No |
| spark.eventLog.logBlockUpdates.enabled | ブロック更新ログの有効化 | No |
| spark.eventLog.logStageExecutorMetrics | ステージ Executor メトリクスログの有効化 | No |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| 1 | イベント発生時刻 | 昇順（時系列順） |

### 改ページ条件

該当なし（ストリーム出力のため改ページ概念なし）

## データベース参照仕様

### 参照テーブル一覧

本帳票はデータベースを参照せず、Spark 内部の SparkListener イベントバスから直接イベントデータを受け取る。

| テーブル名 | 用途 | 結合条件 |
|-----------|------|---------|
| 該当なし | - | - |

### テーブル別参照項目詳細

該当なし（インメモリイベントバスからのデータ取得）

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| ステージ Executor メトリクスピーク | 各 Executor のメトリクス値の最大値（compareAndUpdatePeakValues） | なし | ステージ完了時にログ出力 |
| リダクト済みプロパティ | Utils.redact(sparkConf, properties) | なし | 機密情報を "********" に置換 |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[SparkContext 初期化] --> B{spark.eventLog.enabled?}
    B -->|true| C[EventLoggingListener 生成]
    B -->|false| Z[イベントログなし]
    C --> D[EventLogFileWriter 生成]
    D --> E[ログファイル作成・LogStart 書き込み]
    E --> F[SparkListenerBus に登録]
    F --> G[イベント受信ループ]
    G --> H{イベント種別判定}
    H -->|Stage/Task 系| I[logEvent - フラッシュなし]
    H -->|Job/Executor/App 系| J[logEvent - フラッシュあり]
    H -->|環境情報| K[リダクト処理後 logEvent]
    H -->|ブロック更新| L{logBlockUpdates 有効?}
    L -->|Yes| J
    L -->|No| G
    I --> M[JSON シリアライズ・書き込み]
    J --> M
    K --> M
    M --> G
    G --> N[アプリケーション終了]
    N --> O[logWriter.stop]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| ログディレクトリ不存在 | spark.eventLog.dir で指定されたパスが存在しない | IOException | ディレクトリを事前に作成する |
| 書き込み権限エラー | ログディレクトリへの書き込み権限がない | IOException | 適切なファイルシステム権限を設定する |
| ストレージ容量不足 | 出力先ストレージの容量が不足 | IOException | ストレージ容量を確保するか、古いログを削除する |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | アプリケーションのタスク数に比例（数千〜数百万イベント） |
| 目標出力時間 | イベント発生から書き込みまで数ミリ秒以内 |
| 同時出力数上限 | 1アプリケーションにつき1ファイル |

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

- Spark の設定プロパティに含まれる機密情報（パスワード、トークン等）は `Utils.redact()` によりリダクト処理される（行285-299）
- 環境情報更新イベントでは、"Classpath Entries" 以外の全プロパティセクションがリダクト対象（行307-323）
- ローカルプロパティ（ステージ/ジョブ記述等のカスタムプロパティ）はリダクト対象外

## 備考

- デフォルトログディレクトリは `/tmp/spark-events`（EventLoggingListener.DEFAULT_LOG_DIR）
- テストモード（`spark.eventLog.testing`）では、loggedEvents バッファにもイベントが記録される
- ドライバーの Executor メトリクス更新はダミーステージキー `(-1, -1)` を使用し、全アクティブステージに対して記録される
- EventLogFileWriter は単一ファイルまたはローリング形式での出力をサポートする

---

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

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

### 推奨読解順序

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

SparkListener イベントの型階層と JSON シリアライズの仕組みを理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | SparkListenerEvent の各サブクラス（SparkListenerJobStart, SparkListenerTaskEnd 等）のデータ構造を理解する |
| 1-2 | JsonProtocol.scala | `core/src/main/scala/org/apache/spark/util/JsonProtocol.scala` | sparkEventToJsonString メソッドによるイベントの JSON シリアライズ処理を理解する |

**読解のコツ**: SparkListenerEvent は sealed trait ではないため、各イベントクラスは個別の case class として定義されている。JsonProtocol はパターンマッチでイベント種別を判定し、対応する JSON 変換を行う。

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

EventLoggingListener のライフサイクル（生成・開始・停止）を把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | クラス定義（行48-54）でコンストラクタパラメータを確認。start() メソッド（行82-85）でログファイル初期化処理を理解する |

**主要処理フロー**:
1. **行48-54**: コンストラクタ - appId, logBaseDir, sparkConf 等を受け取る
2. **行63-64**: EventLogFileWriter の生成
3. **行82-85**: start() - ログファイル作成と LogStart メタデータ書き込み
4. **行87-93**: initEventLog() - SPARK_VERSION を含む LogStart イベントを出力
5. **行97-103**: logEvent() - 各イベントを JSON シリアライズしてログファイルに書き込む

#### Step 3: イベントハンドリング層を理解する

各 SparkListener メソッドのオーバーライドとフラッシュ制御を把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | 行106-278 の各 override メソッド。フラッシュあり/なしの使い分けに注目 |

**主要処理フロー**:
- **行106-113**: onStageSubmitted - フラッシュなし、プロパティリダクト処理、ステージ Executor メトリクス追跡開始
- **行119-129**: onTaskEnd - フラッシュなし、Executor メトリクスピーク値更新
- **行136-158**: onStageCompleted - フラッシュあり、Executor メトリクスピーク値をログ出力、前回試行のメトリクスをクリア
- **行160-164**: onJobStart/End - フラッシュあり、プロパティリダクト処理
- **行250-268**: onExecutorMetricsUpdate - ドライバーの場合は全ステージに対してピーク値を更新

#### Step 4: リダクト処理を理解する

機密情報の保護メカニズムを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | 行285-299 の redactProperties メソッド、行307-323 の redactEvent メソッド |

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

```
SparkContext
    |
    +-- EventLoggingListener (SparkListener)
    |       |
    |       +-- start()
    |       |     +-- logWriter.start()
    |       |     +-- initEventLog()
    |       |           +-- JsonProtocol.sparkEventToJsonString()
    |       |           +-- logWriter.writeEvent()
    |       |
    |       +-- onStageSubmitted() / onTaskEnd() / onJobStart() / ...
    |       |     +-- redactProperties()
    |       |     |     +-- Utils.redact()
    |       |     +-- logEvent()
    |       |           +-- JsonProtocol.sparkEventToJsonString()
    |       |           +-- logWriter.writeEvent()
    |       |
    |       +-- onStageCompleted()
    |       |     +-- liveStageExecutorMetrics 処理
    |       |     +-- logEvent(SparkListenerStageExecutorMetrics)
    |       |     +-- logEvent(event, flushLogger=true)
    |       |
    |       +-- stop()
    |             +-- logWriter.stop()
    |
    +-- EventLogFileWriter
            +-- writeEvent()
            +-- start() / stop()
```

### データフロー図

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

SparkListenerBus     ──▶  EventLoggingListener           ──▶  イベントログファイル
 (イベント発生)              |                                   (JSON Lines)
                            +-- JsonProtocol
                            |   (JSON シリアライズ)          ──▶  HDFS/S3/ローカルFS
                            +-- Utils.redact                     ({appId}_{attemptId})
                            |   (機密情報リダクト)
                            +-- EventLogFileWriter
                                (ファイル書き込み)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.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 シリアライズ/デシリアライズ |
| EventLogFileWriter.scala | `core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriter.scala` | ソース | ログファイルの物理的な書き込み処理 |
| SparkContext.scala | `core/src/main/scala/org/apache/spark/SparkContext.scala` | ソース | EventLoggingListener の生成と登録 |
| Utils.scala | `core/src/main/scala/org/apache/spark/util/Utils.scala` | ソース | redact メソッドによる機密情報保護 |
| ExecutorMetrics.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorMetrics.scala` | ソース | Executor メトリクスのピーク値管理 |
