# 帳票設計書 10-StatsReportListener 統計サマリー

## 概要

本ドキュメントは、Apache Spark のスケジューラにおける StatsReportListener が出力する統計サマリーレポートの設計書である。SparkListener インターフェースを実装し、各ステージの完了時にタスクのランタイム分布、シャッフルバイト数、フェッチ待ち時間等の統計情報をログ出力する。Dropwizard MetricRegistry ベースの Sink とは異なり、SparkListener イベント駆動型の統計レポートである。

### 本帳票の処理概要

StatsReportListener 統計サマリーは、Spark ジョブの各ステージ完了時にタスクレベルの実行統計をパーセンタイル分布として算出し、Logging フレームワーク経由でログ出力する帳票機能である。

**業務上の目的・背景**：Spark アプリケーションのパフォーマンスチューニングにおいて、タスクレベルの実行時間分布やシャッフルI/O特性を把握することが不可欠である。本リスナーはステージ完了時に自動的に統計サマリーを出力することで、パフォーマンスボトルネックの特定を支援する。特にデータスキュー（タスク間の実行時間の偏り）の検出に有効であり、0%, 5%, 10%, 25%, 50%, 75%, 90%, 95%, 100% のパーセンタイル分布を提供する。

**帳票の利用シーン**：(1) Spark アプリケーションのパフォーマンスチューニング時にタスク実行時間分布を分析する場合、(2) データスキューの有無を確認する場合、(3) シャッフル I/O のボトルネックを特定する場合、(4) ステージレベルの詳細な実行統計を取得する場合。

**主要な出力内容**：
1. タスクランタイム分布（ミリ秒表記）- 各タスクの実行時間のパーセンタイル分布
2. シャッフル書き込みバイト分布 - 各タスクが書き込んだシャッフルデータ量の分布
3. フェッチ待ち時間分布 - シャッフルデータ取得時の待ち時間の分布
4. リモート読み取りバイト分布 - リモートノードから読み取ったデータ量の分布
5. タスク結果サイズ分布 - タスク結果のサイズ分布
6. ランタイム内訳（Executor 非フェッチ時間、フェッチ待ち時間、その他）のパーセンテージ分布

**帳票の出力タイミング**：各ステージが完了した時点（onStageCompleted イベント発火時）で自動的に出力される。タスク終了時（onTaskEnd）にタスク情報を蓄積し、ステージ完了時に集計・出力・クリアする。

**帳票の利用者**：Spark アプリケーション開発者、パフォーマンスエンジニア、データエンジニア。

## 帳票種別

統計レポート（イベント駆動型のステージ完了時統計サマリー）

## 利用画面

| 画面No | 画面名 | URL/ルーティング | 出力操作 |
|--------|--------|-----------------|---------|
| N/A | SparkContext への登録 | N/A | `sc.addSparkListener(new StatsReportListener())` でプログラム的に登録 |

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | テキスト（Logging フレームワーク経由） |
| 用紙サイズ | N/A |
| 向き | N/A |
| ファイル名 | ログ設定に依存 |
| 出力方法 | INFO レベルのログ出力 |
| 文字コード | ログ設定に依存 |

### 統計レポート固有設定

| 項目 | 内容 |
|-----|------|
| パーセンタイル | 0%, 5%, 10%, 25%, 50%, 75%, 90%, 95%, 100% |
| 時間表示 | 自動単位変換（ms/s/min/hours） |
| バイト表示 | 自動単位変換（Utils.bytesToString） |

## 帳票レイアウト

### レイアウト概要

ステージ完了ごとに以下の構造でログ出力される。

```
Finished stage: Stage({stageId}, {attemptNumber}); Name: '{name}'; Status: {status}; numTasks: {n}; Took: {ms} msec
task runtime:    {stats}
                 0%     5%     10%    25%    50%    75%    90%    95%    100%
                 {v}    {v}    {v}    {v}    {v}    {v}    {v}    {v}    {v}
shuffle bytes written:   {stats}
                 0%     5%     10%    25%    50%    75%    90%    95%    100%
                 {v}    {v}    {v}    {v}    {v}    {v}    {v}    {v}    {v}
fetch wait time: ...
remote bytes read: ...
task result size: ...
executor (non-fetch) time pct: ...
fetch wait time pct: ...
other time pct: ...
```

### ヘッダー部

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | ステージ情報 | ステージID、試行番号、名前、ステータス、タスク数、所要時間 | SparkListenerStageCompleted.stageInfo | "Stage({id}, {attempt}); Name: '{name}'; Status: {status}; numTasks: {n}; Took: {ms} msec" |

### 明細部

| No | 項目名 | 説明 | データ取得元 | 表示形式 | 列幅 |
|----|-------|------|-------------|---------|-----|
| 1 | task runtime | タスク実行時間分布 | TaskInfo.duration | ミリ秒→自動単位変換 | 可変 |
| 2 | shuffle bytes written | シャッフル書き込みバイト分布 | ShuffleWriteMetrics.bytesWritten | バイト→自動単位変換 | 可変 |
| 3 | fetch wait time | フェッチ待ち時間分布 | ShuffleReadMetrics.fetchWaitTime | ミリ秒→自動単位変換 | 可変 |
| 4 | remote bytes read | リモート読み取りバイト分布 | ShuffleReadMetrics.remoteBytesRead | バイト→自動単位変換 | 可変 |
| 5 | task result size | タスク結果サイズ分布 | TaskMetrics.resultSize | バイト→自動単位変換 | 可変 |
| 6 | executor (non-fetch) time pct | Executor 非フェッチ時間の割合 | RuntimePercentage.executorPct | "%2.0f %%" | 可変 |
| 7 | fetch wait time pct | フェッチ待ち時間の割合 | RuntimePercentage.fetchPct | "%2.0f %%" | 可変 |
| 8 | other time pct | その他時間の割合 | RuntimePercentage.other | "%2.0f %%" | 可変 |

### フッター部

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| N/A | フッターなし | N/A | N/A | N/A |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| ステージ完了イベント | SparkListenerStageCompleted イベントの発火 | Yes |
| タスク情報の蓄積 | onTaskEnd で蓄積された TaskInfo と TaskMetrics のペア（null でないもの） | Yes |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| 1 | パーセンタイル | 固定順（0%, 5%, 10%, 25%, 50%, 75%, 90%, 95%, 100%） |

### 改ページ条件

ログ出力のため改ページなし。ステージ完了ごとに出力される。

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

### 参照テーブル一覧

本帳票はデータベースを参照しない。SparkListener イベントから直接データを取得する。

| テーブル名 | 用途 | 結合条件 |
|-----------|------|---------|
| N/A（SparkListener イベント） | タスク実行情報の取得元 | N/A |

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

#### TaskInfo（インメモリ）

| 参照項目 | 帳票項目との対応 | 取得条件 | 備考 |
|---------|----------------|---------|------|
| duration | task runtime 分布 | onTaskEnd で蓄積 | ミリ秒 |

#### TaskMetrics（インメモリ）

| 参照項目 | 帳票項目との対応 | 取得条件 | 備考 |
|---------|----------------|---------|------|
| shuffleWriteMetrics.bytesWritten | shuffle bytes written 分布 | onTaskEnd で蓄積 | バイト |
| shuffleReadMetrics.fetchWaitTime | fetch wait time 分布 | onTaskEnd で蓄積 | ミリ秒 |
| shuffleReadMetrics.remoteBytesRead | remote bytes read 分布 | onTaskEnd で蓄積 | バイト |
| resultSize | task result size 分布 | onTaskEnd で蓄積 | バイト |
| executorRunTime | RuntimePercentage 計算 | onTaskEnd で蓄積 | ミリ秒 |

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| パーセンタイル分布 | Distribution.getQuantiles(probabilities) | N/A | probabilities = [0.0, 0.05, 0.1, 0.25, 0.5, 0.75, 0.9, 0.95, 1.0] |
| executorPct | (executorRunTime - fetchWaitTime) / totalTime | 100倍して %2.0f %% | RuntimePercentage.apply |
| fetchPct | fetchWaitTime / totalTime | 100倍して %2.0f %% | Option 型（fetchTime が存在する場合のみ） |
| other | 1.0 - (executorPct + fetchPct) | 100倍して %2.0f %% | 残余時間の割合 |
| millisToString | ms > hours: hours 単位、ms > minutes: min 単位、ms > seconds: s 単位、otherwise: ms | "%.1f %s" | 自動単位変換 |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[タスク完了 - onTaskEnd] --> B{info != null && metrics != null?}
    B -->|Yes| C[taskInfoMetrics バッファに追加]
    B -->|No| D[スキップ]
    E[ステージ完了 - onStageCompleted] --> F[ステージ情報をログ出力]
    F --> G[task runtime 分布を出力]
    G --> H[shuffle bytes written 分布を出力]
    H --> I[fetch wait time 分布を出力]
    I --> J[remote bytes read 分布を出力]
    J --> K[task result size 分布を出力]
    K --> L[RuntimePercentage 計算]
    L --> M[executor/fetch/other 時間割合を出力]
    M --> N[taskInfoMetrics バッファをクリア]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| タスク情報なし | TaskInfo または TaskMetrics が null | 出力なし（バッファに追加されない） | 正常動作、対処不要 |
| 分布計算不可 | タスクが0件 | 分布が None になり出力なし | 正常動作、対処不要 |
| ゼロ除算 | totalTime が 0 | RuntimePercentage の計算で NaN/Infinity | ステージに実行タスクがない場合に発生する可能性あり |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | ステージ内のタスク数に依存（数千〜数万タスク） |
| 目標出力時間 | ステージ完了イベント処理内で完了（通常数十ミリ秒以内） |
| 同時出力数上限 | 1（SparkListener のイベント処理は逐次） |

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

- ログ出力にはステージ名やタスク実行統計が含まれるが、機密データは含まれない
- ログファイルのアクセス権限はログ設定に依存する

## 備考

- `@DeveloperApi` アノテーションが付与されている
- SparkContext に登録するにはプログラムコードが必要：
  ```scala
  sc.addSparkListener(new StatsReportListener())
  ```
  または `spark.extraListeners` 設定で指定：
  ```
  spark.extraListeners=org.apache.spark.scheduler.StatsReportListener
  ```
- ステージ完了ごとに taskInfoMetrics バッファがクリアされるため、メモリ使用量は増大しない
- Distribution ユーティリティクラスを使用してパーセンタイル計算を行う
- 時間の自動単位変換: 1h 以上 -> hours, 1min 以上 -> min, 1s 以上 -> s, otherwise -> ms
- RuntimePercentage は executorPct + fetchPct + other = 1.0 となるように計算される

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StatsReportListener.scala | `core/src/main/scala/org/apache/spark/scheduler/StatsReportListener.scala` | RuntimePercentage case class（行189-199）を確認。executorPct, fetchPct(Option), other の3つの割合フィールドを持つ |

**読解のコツ**: RuntimePercentage はコンパニオンオブジェクトの apply メソッド（行192-199）で生成される。totalTime に対する各時間の割合を計算し、残余を other に割り当てる。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | StatsReportListener.scala | `core/src/main/scala/org/apache/spark/scheduler/StatsReportListener.scala` | StatsReportListener クラス（行33-88）の onTaskEnd と onStageCompleted を確認 |

**主要処理フロー**:
1. **行37**: taskInfoMetrics バッファ定義（mutable.Buffer[(TaskInfo, TaskMetrics)]）
2. **行39-45**: onTaskEnd() - info と metrics が null でなければバッファに追加
3. **行47-75**: onStageCompleted() - メインのレポート出力ロジック
   - **行49-50**: ステージ情報のログ出力（getStatusDetail）
   - **行51**: task runtime 分布出力（showMillisDistribution）
   - **行54-55**: shuffle bytes written 分布出力（showBytesDistribution）
   - **行58-59**: fetch wait time 分布出力（showMillisDistribution）
   - **行60-61**: remote bytes read 分布出力（showBytesDistribution）
   - **行62-63**: task result size 分布出力（showBytesDistribution）
   - **行66-68**: RuntimePercentage の計算
   - **行69-73**: executor/fetch/other 時間割合の分布出力
   - **行74**: taskInfoMetrics バッファのクリア

#### Step 3: コンパニオンオブジェクトのユーティリティを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StatsReportListener.scala | `core/src/main/scala/org/apache/spark/scheduler/StatsReportListener.scala` | コンパニオンオブジェクト（行90-187）の統計ユーティリティメソッド群 |

**主要処理フロー**:
- **行93-95**: percentiles 配列 = [0, 5, 10, 25, 50, 75, 90, 95, 100]、probabilities = percentiles / 100.0
- **行97-110**: extractDoubleDistribution/extractLongDistribution - タスクメトリクスから Distribution を抽出
- **行112-138**: showDistribution の各オーバーロード - Distribution のパーセンタイルをフォーマットしてログ出力
- **行140-165**: showBytesDistribution/showMillisDistribution - バイト/ミリ秒用のフォーマッタ付き分布出力
- **行167-186**: millisToString - ミリ秒を適切な単位に自動変換（hours/min/s/ms）

#### Step 4: RuntimePercentage の計算ロジックを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | StatsReportListener.scala | `core/src/main/scala/org/apache/spark/scheduler/StatsReportListener.scala` | RuntimePercentage オブジェクト（行191-199）の apply メソッド |

**主要処理フロー**:
- **行192**: totalTime を Double に変換（denom）
- **行194**: fetchTime = Some(shuffleReadMetrics.fetchWaitTime)
- **行195**: fetch = fetchTime / denom
- **行196**: exec = (executorRunTime - fetchTime) / denom
- **行197**: other = 1.0 - (exec + fetch)
- **行198**: RuntimePercentage(exec, fetch, other) を返す

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

```
SparkContext.addSparkListener(StatsReportListener)
    |
    +-- [タスク完了イベント]
    |       +-- onTaskEnd(SparkListenerTaskEnd)
    |               +-- taskInfoMetrics += (info, metrics)
    |
    +-- [ステージ完了イベント]
            +-- onStageCompleted(SparkListenerStageCompleted)
                    +-- getStatusDetail(stageInfo) -> ステージ情報ログ
                    +-- showMillisDistribution("task runtime:", ...)
                    |       +-- extractLongDistribution(taskInfoMetrics, info.duration)
                    |       +-- Distribution.getQuantiles(probabilities)
                    |       +-- millisToString(ms)
                    |       +-- logInfo(...)
                    +-- showBytesDistribution("shuffle bytes written:", ...)
                    |       +-- extractLongDistribution(taskInfoMetrics, metrics.shuffleWriteMetrics.bytesWritten)
                    |       +-- Utils.bytesToString(d.toLong)
                    +-- showMillisDistribution("fetch wait time:", ...)
                    +-- showBytesDistribution("remote bytes read:", ...)
                    +-- showBytesDistribution("task result size:", ...)
                    +-- RuntimePercentage.apply(info.duration, metrics)
                    +-- showDistribution("executor (non-fetch) time pct:", ...)
                    +-- showDistribution("fetch wait time pct:", ...)
                    +-- showDistribution("other time pct:", ...)
                    +-- taskInfoMetrics.clear()
```

### データフロー図

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

SparkListenerTaskEnd        StatsReportListener            ログ出力 (INFO)
+-- TaskInfo                +-- onTaskEnd()                 Finished stage: ...
+-- TaskMetrics                 +-- バッファに蓄積           task runtime: {stats}
                                                             0%  5%  10%  ...
SparkListenerStageCompleted +-- onStageCompleted()           {v} {v} {v}  ...
+-- StageInfo                   +-- 統計計算                 shuffle bytes written: ...
                                +-- パーセンタイル算出       fetch wait time: ...
                                +-- RuntimePercentage       remote bytes read: ...
                                +-- ログ出力                task result size: ...
                                +-- バッファクリア          executor time pct: ...
                                                            fetch wait time pct: ...
                                                            other time pct: ...
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StatsReportListener.scala | `core/src/main/scala/org/apache/spark/scheduler/StatsReportListener.scala` | ソース | 統計サマリーレポートのリスナー実装 |
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | SparkListener トレイト定義 |
| TaskInfo.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala` | ソース | タスク情報データクラス（duration 等） |
| TaskMetrics.scala | `core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala` | ソース | タスクメトリクスデータクラス |
| Distribution.scala | `core/src/main/scala/org/apache/spark/util/Distribution.scala` | ソース | 分布計算ユーティリティ（パーセンタイル算出） |
| Utils.scala | `core/src/main/scala/org/apache/spark/util/Utils.scala` | ソース | bytesToString 等のユーティリティ |
