# 帳票設計書 1-CsvSink メトリクスレポート

## 概要

本ドキュメントは、Apache Spark の MetricsSystem における CsvSink が出力する CSV 形式メトリクスレポートの設計書である。Dropwizard Metrics ライブラリの CsvReporter を利用し、Spark の各コンポーネント（Driver、Executor、Master、Worker 等）のメトリクス情報を定期的に CSV ファイルとしてローカルファイルシステムへ出力する。

### 本帳票の処理概要

CsvSink メトリクスレポートは、Spark の内部メトリクスデータを CSV ファイルとして定期出力する帳票機能である。

**業務上の目的・背景**：Spark クラスタの運用監視において、メトリクス情報を永続化し、後から時系列分析を行う必要がある。CSV 形式は汎用的なフォーマットであり、Excel やデータ分析ツールで容易に読み込めるため、運用チームが独自の分析基盤を構築する際に利便性が高い。外部モニタリングシステム（Graphite、Prometheus 等）を導入していない環境でも、ファイルベースでメトリクスを蓄積できる軽量な監視手段を提供する。

**帳票の利用シーン**：(1) Spark アプリケーションのパフォーマンス分析を後日実施する場合、(2) ネットワーク接続が制限された環境でメトリクスを蓄積したい場合、(3) メトリクスデータを独自の BI ツールや分析パイプラインに投入する場合に利用される。

**主要な出力内容**：
1. Gauge メトリクス（JVM メモリ使用量、アクティブタスク数等）の時系列値
2. Counter メトリクス（完了タスク数、失敗タスク数等）の累積値
3. Histogram メトリクス（タスク実行時間分布等）の統計値（min, max, mean, 各パーセンタイル）
4. Meter メトリクス（スループット等）のレート情報（1分/5分/15分移動平均）
5. Timer メトリクス（API 応答時間等）の統計値とレート情報

**帳票の出力タイミング**：MetricsSystem の start() 呼び出し時に定期出力が開始され、設定された間隔（デフォルト10秒）で自動的に CSV ファイルが更新される。また report() メソッドにより手動での即時出力も可能。

**帳票の利用者**：Spark クラスタ管理者、運用チーム、パフォーマンスエンジニア、SRE チーム。

## 帳票種別

メトリクスレポート（定期出力型の時系列データファイル）

## 利用画面

| 画面No | 画面名 | URL/ルーティング | 出力操作 |
|--------|--------|-----------------|---------|
| N/A | 設定ファイル駆動 | N/A | metrics.properties の設定により自動出力 |

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | CSV |
| 用紙サイズ | N/A（電子ファイル） |
| 向き | N/A |
| ファイル名 | {メトリクス名}.csv（例: `jvm.heap.used.csv`） |
| 出力方法 | ローカルファイルシステムへの定期書き込み |
| 文字コード | UTF-8（US ロケール） |

### CSV 固有設定

| 項目 | 内容 |
|-----|------|
| 区切り文字 | カンマ（Dropwizard CsvReporter のデフォルト） |
| ヘッダー行 | あり（CsvReporter が自動付与） |
| 数値フォーマット | US ロケール（小数点はピリオド） |
| 時間単位（Duration） | ミリ秒（MILLISECONDS） |
| レート単位 | 秒あたり（SECONDS） |

## 帳票レイアウト

### レイアウト概要

各メトリクス種別に対し、CsvReporter がメトリクス名ごとに個別の CSV ファイルを生成する。各ファイルはタイムスタンプ付きの行が追記される。

```
出力ディレクトリ/
  ├── {gauge_name}.csv
  ├── {counter_name}.csv
  ├── {histogram_name}.csv
  ├── {meter_name}.csv
  └── {timer_name}.csv
```

### ヘッダー部

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | t | タイムスタンプ | システム時刻 | エポック秒 |

### 明細部

メトリクス種別により出力カラムが異なる。

**Gauge の場合:**

| No | 項目名 | 説明 | データ取得元 | 表示形式 | 列幅 |
|----|-------|------|-------------|---------|-----|
| 1 | t | タイムスタンプ | システム時刻 | エポック秒 | 可変 |
| 2 | value | メトリクス値 | MetricRegistry.Gauge.getValue | 数値 | 可変 |

**Counter の場合:**

| No | 項目名 | 説明 | データ取得元 | 表示形式 | 列幅 |
|----|-------|------|-------------|---------|-----|
| 1 | t | タイムスタンプ | システム時刻 | エポック秒 | 可変 |
| 2 | count | 累積カウント | MetricRegistry.Counter.getCount | 整数 | 可変 |

**Histogram の場合:**

| No | 項目名 | 説明 | データ取得元 | 表示形式 | 列幅 |
|----|-------|------|-------------|---------|-----|
| 1 | t | タイムスタンプ | システム時刻 | エポック秒 | 可変 |
| 2 | count | サンプル数 | Histogram.getCount | 整数 | 可変 |
| 3 | max | 最大値 | Snapshot.getMax | 数値 | 可変 |
| 4 | mean | 平均値 | Snapshot.getMean | 数値 | 可変 |
| 5 | min | 最小値 | Snapshot.getMin | 数値 | 可変 |
| 6 | stddev | 標準偏差 | Snapshot.getStdDev | 数値 | 可変 |
| 7 | p50 | 中央値 | Snapshot.getMedian | 数値 | 可変 |
| 8 | p75 | 75パーセンタイル | Snapshot.get75thPercentile | 数値 | 可変 |
| 9 | p95 | 95パーセンタイル | Snapshot.get95thPercentile | 数値 | 可変 |
| 10 | p98 | 98パーセンタイル | Snapshot.get98thPercentile | 数値 | 可変 |
| 11 | p99 | 99パーセンタイル | Snapshot.get99thPercentile | 数値 | 可変 |
| 12 | p999 | 99.9パーセンタイル | Snapshot.get999thPercentile | 数値 | 可変 |

### フッター部

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| N/A | フッターなし | CSV ファイルにフッターは存在しない | N/A | N/A |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| MetricRegistry 登録 | MetricRegistry に登録されている全メトリクスが対象 | Yes |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| 1 | タイムスタンプ | 昇順（追記型） |

### 改ページ条件

CSV ファイルのため改ページなし。メトリクス名ごとに個別ファイルとして出力される。

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

### 参照テーブル一覧

本帳票はデータベースを参照しない。Dropwizard MetricRegistry のインメモリデータを直接参照する。

| テーブル名 | 用途 | 結合条件 |
|-----------|------|---------|
| N/A（MetricRegistry） | メトリクスデータの取得元 | N/A |

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

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

| 参照項目 | 帳票項目との対応 | 取得条件 | 備考 |
|---------|----------------|---------|------|
| Gauges | gauge CSV ファイル | registry.getGauges() | 全 Gauge を出力 |
| Counters | counter CSV ファイル | registry.getCounters() | 全 Counter を出力 |
| Histograms | histogram CSV ファイル | registry.getHistograms() | 全 Histogram を出力 |
| Meters | meter CSV ファイル | registry.getMeters() | 全 Meter を出力 |
| Timers | timer CSV ファイル | registry.getTimers() | 全 Timer を出力 |

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| レート変換 | 元値 / 秒 | Dropwizard 既定 | convertRatesTo(TimeUnit.SECONDS) |
| Duration変換 | 元値(ns) -> ms | Dropwizard 既定 | convertDurationsTo(TimeUnit.MILLISECONDS) |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[MetricsSystem.start] --> B[CsvSink.start]
    B --> C[CsvReporter.start - 定期スケジュール登録]
    C --> D{定期タイマー発火}
    D --> E[CsvReporter.report]
    E --> F[MetricRegistry からメトリクス取得]
    F --> G[メトリクス種別ごとに CSV ファイルへ追記]
    G --> D
    H[MetricsSystem.stop] --> I[CsvSink.stop]
    I --> J[CsvReporter.stop - スケジュール停止]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| ポーリング間隔エラー | pollUnit と pollPeriod の組み合わせが最小ポーリング間隔（1秒）未満 | IllegalArgumentException: Polling period ... below than minimal polling period | 設定値を1秒以上に変更 |
| ディレクトリアクセスエラー | 出力先ディレクトリが存在しないまたは書き込み権限がない | IOException（CsvReporter 内部） | ディレクトリの作成・権限設定 |
| 数値変換エラー | period 設定値が数値でない | NumberFormatException | 整数値を設定 |
| TimeUnit 変換エラー | unit 設定値が不正な TimeUnit 名 | IllegalArgumentException | SECONDS, MINUTES 等の有効な値を設定 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | メトリクス数に依存（通常数十〜数百メトリクス） |
| 目標出力時間 | 定期間隔内に完了（デフォルト10秒以内） |
| 同時出力数上限 | 1（単一スレッドでの定期レポート） |

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

- 出力先ディレクトリのファイルシステム権限により、メトリクスデータへのアクセスが制御される
- デフォルト出力先が `/tmp/` であるため、本番環境では適切な権限を持つディレクトリに変更すべき
- メトリクスデータにはクラスタの内部状態が含まれるため、外部からアクセスできない場所に保存すること

## 備考

- 設定は `metrics.properties` ファイルで行う。設定例：
  ```
  *.sink.csv.class=org.apache.spark.metrics.sink.CsvSink
  *.sink.csv.period=10
  *.sink.csv.unit=SECONDS
  *.sink.csv.directory=/var/spark/metrics/
  ```
- CsvReporter は Dropwizard Metrics ライブラリ（com.codahale.metrics）の機能を利用している
- メトリクス名ごとに個別の CSV ファイルが生成されるため、メトリクス数が多い場合はファイル数が増大する

---

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

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

### 推奨読解順序

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

まず、Sink インターフェースの構造と MetricRegistry のデータモデルを理解することが重要である。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Sink.scala | `core/src/main/scala/org/apache/spark/metrics/sink/Sink.scala` | Sink トレイトの3メソッド（start/stop/report）を確認。全 Sink 共通のインターフェース定義（行20-24） |

**読解のコツ**: Scala の trait は Java のインターフェースに相当する。`private[spark]` はパッケージプライベートを意味する。

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

MetricsSystem がどのように Sink を生成・管理しているかを把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | MetricsSystem.scala | `core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala` | registerSinks() メソッド（行199-240）で Sink のインスタンス化とレジストリ登録を確認。start() メソッド（行97-106）で sinks.foreach(_.start()) により全 Sink が起動される流れを理解する |

**主要処理フロー**:
1. **行97-106**: MetricsSystem.start() が registerSinks() を呼び出し、設定に基づき Sink をインスタンス化後、全 Sink の start() を呼び出す
2. **行199-240**: registerSinks() が metrics.properties の設定を読み込み、リフレクションで Sink クラスをインスタンス化する
3. **行252-258**: checkMinimalPollingPeriod() でポーリング間隔の最小値（1秒）を検証する

#### Step 3: CsvSink の実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | CsvSink.scala | `core/src/main/scala/org/apache/spark/metrics/sink/CsvSink.scala` | クラス全体（行28-72）を通読。コンストラクタで設定値の読み込み、CsvReporter の構築、start/stop/report メソッドの実装を確認する |

**主要処理フロー**:
- **行28-29**: Properties と MetricRegistry をコンストラクタ引数として受け取る
- **行30-36**: 設定キーとデフォルト値の定義（period=10, unit=SECONDS, directory=/tmp/）
- **行38-53**: Properties から設定値を読み込み、デフォルト値でフォールバック
- **行48**: MetricsSystem.checkMinimalPollingPeriod() でポーリング間隔を検証
- **行55-59**: CsvReporter.forRegistry() でレポーターを構築。US ロケール、Duration はミリ秒、Rate は秒単位
- **行61-63**: start() で CsvReporter の定期レポートを開始
- **行65-67**: stop() で定期レポートを停止
- **行69-71**: report() で即時レポートを実行

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

```
MetricsSystem.start()
    |
    +-- registerSinks()
    |       +-- Utils.classForName("CsvSink")
    |       +-- CsvSink.<init>(properties, registry)
    |               +-- MetricsSystem.checkMinimalPollingPeriod()
    |               +-- CsvReporter.forRegistry(registry).build(pollDir)
    |
    +-- sinks.foreach(_.start())
            +-- CsvSink.start()
                    +-- CsvReporter.start(pollPeriod, pollUnit)
                            +-- [定期スケジュール] CsvReporter.report()
                                    +-- MetricRegistry.getGauges/getCounters/...
                                    +-- CSV ファイル書き込み
```

### データフロー図

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

MetricRegistry         CsvSink                     CSV ファイル群
(Gauges,               +-- CsvReporter              /var/spark/metrics/
 Counters,                 +-- forRegistry()         +-- gauge1.csv
 Histograms,               +-- formatFor(US)         +-- counter1.csv
 Meters,                   +-- convertDurations(ms)  +-- histogram1.csv
 Timers)                   +-- convertRates(/s)      +-- meter1.csv
                           +-- build(pollDir)        +-- timer1.csv

Properties             設定読み込み
(period, unit, dir)    +-- pollPeriod, pollUnit, pollDir
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Sink.scala | `core/src/main/scala/org/apache/spark/metrics/sink/Sink.scala` | ソース | Sink トレイト定義（start/stop/report） |
| CsvSink.scala | `core/src/main/scala/org/apache/spark/metrics/sink/CsvSink.scala` | ソース | CSV 出力 Sink の実装 |
| MetricsSystem.scala | `core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala` | ソース | Sink の登録・管理・起動を行うシステムクラス |
| MetricsConfig.scala | `core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala` | ソース | metrics.properties の読み込み・パース |
| metrics.properties | `conf/metrics.properties.template` | 設定 | メトリクス出力の設定ファイルテンプレート |
