# 機能設計書 97-Code Profiler

## 概要

本ドキュメントは、Apache SparkのCode Profiler機能の設計について記述する。Async Profilerを利用してドライバーおよびエグゼキューターのJFRプロファイリングを行い、結果をHDFS互換ファイルシステムに保存する機能を提供する。

### 本機能の処理概要

**業務上の目的・背景**：Sparkアプリケーションのパフォーマンス分析において、CPU使用状況、メモリ割り当て、ロック競合などの低レベルプロファイリング情報は問題特定に不可欠である。本機能はSparkプラグインフレームワークを利用して、Async Profilerを透過的にドライバー/エグゼキューターに統合し、プロファイリングデータを自動的にDFS上に収集する。

**機能の利用シーン**：Sparkアプリケーションのパフォーマンス調査・チューニングにおいて、CPU、メモリ割り当て、ロック競合等のプロファイリングを実施する場面で利用される。

**主要な処理内容**：
1. ProfilerPluginによるSparkプラグインフレームワークへの統合（DriverPlugin/ExecutorPlugin）
2. SparkAsyncProfilerによるAsync Profilerのラップと制御（start/stop/dump/resume）
3. 定期的なDFS書き込み（dump→アップロード→resume）
4. エグゼキューターのサンプリング（fraction設定による確率的プロファイリング対象選択）
5. JFR形式のプロファイリング出力

**関連システム・外部連携**：Async Profiler（ネイティブライブラリ）、HDFS互換ファイルシステム（プロファイル出力先）、Sparkプラグインフレームワーク（SparkPlugin API）。

**権限による制御**：DFS出力ディレクトリのパーミッション管理（770/660）。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 画面機能マッピングに本機能に対応する画面定義なし |

## 機能種別

ユーティリティ / パフォーマンスプロファイリング

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| spark.profiler.driver.enabled | boolean | No | ドライバープロファイリング有効化（デフォルト: false） | - |
| spark.profiler.executor.enabled | boolean | No | エグゼキュータープロファイリング有効化（デフォルト: false） | - |
| spark.profiler.executor.fraction | double | No | プロファイリング対象エグゼキューター割合（デフォルト: 0.1） | [0.0, 1.0]範囲 |
| spark.profiler.dfsDir | String | No | DFS出力ディレクトリ | HDFS互換パス |
| spark.profiler.localDir | String | No | ローカル一時ディレクトリ（デフォルト: "."） | 有効なパス |
| spark.profiler.asyncProfiler.args | String | No | Async Profilerの引数 | デフォルト: "event=wall,interval=10ms,alloc=2m,lock=10ms,chunktime=300s" |
| spark.profiler.dfsWriteInterval | long | No | DFS書き込み間隔（デフォルト: 30秒） | 非負値 |

### 入力データソース

Async Profilerが生成するローカルJFRファイル

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| プロファイルファイル | JFR | profile-driver.jfr / profile-exec-{id}.jfr |

### 出力先

- ローカル: `${localDir}/profile-{driver|exec-{id}}.jfr`
- DFS: `${dfsDir}/${appId}/${profileFile}`

### ファイルパーミッション

| 対象 | パーミッション |
|-----|-------------|
| DFS出力ディレクトリ | 770 |
| DFS出力ファイル | 660 |

## 処理フロー

### 処理シーケンス

```
1. プラグイン初期化
   a. ドライバー側（ProfilerDriverPlugin.init）:
      - PROFILER_DRIVER_ENABLED確認
      - SparkAsyncProfiler生成（executorID = "driver"）
      - profiler.start()
   b. エグゼキューター側（ProfilerExecutorPlugin.init）:
      - PROFILER_EXECUTOR_ENABLED確認
      - PROFILER_EXECUTOR_FRACTIONに基づくサンプリング判定
      - rand.nextInt(100) * 0.01 < fraction の場合のみ実行
      - SparkAsyncProfiler生成（executorID = actual executor ID）
      - profiler.start()

2. プロファイリング開始（SparkAsyncProfiler.start）
   a. AsyncProfilerLoader.isSupported確認
   b. AsyncProfilerLoader.load()でネイティブライブラリロード
   c. profiler.execute(startcmd): "start,{options},file={localDir}/{profileFile}"
   d. startWriting(): DFS書き込みスレッド開始

3. 定期的なDFS書き込み（startWriting）
   a. InputStreamでローカルJFRファイルを開く
   b. ScheduledExecutorServiceで定期タスク起動（writeInterval秒間隔）
   c. writeChunk(false):
      - profiler.execute(stopcmd): 一時停止
      - profiler.execute(dumpcmd): ダンプ
      - profiler.execute(resumecmd): 再開
      - inputStream.read → outputStream.write（8MBバッファ）
   d. 初回writeChunk時にDFS出力先を決定:
      - appIdの取得を待機
      - ${dfsDir}/${appId_attemptId}/ ディレクトリ作成（770）
      - FSDataOutputStream作成（660）

4. プロファイリング停止（SparkAsyncProfiler.stop）
   a. profiler.execute(stopcmd): "stop,{options},file={localDir}/{profileFile}"
   b. finishWriting():
      - threadpool.shutdown() + awaitTermination(30秒)
      - writeChunk(true): 最終データフラッシュ
      - inputStream.close() + outputStream.close()

5. プラグインシャットダウン
   a. ProfilerDriverPlugin.shutdown() / ProfilerExecutorPlugin.shutdown()
   b. profiler.stop()
```

### フローチャート

```mermaid
flowchart TD
    A[Sparkプラグイン初期化] --> B{ドライバー/エグゼキューター?}
    B -->|ドライバー| C{driver.enabled?}
    B -->|エグゼキューター| D{executor.enabled?}
    C -->|Yes| E[SparkAsyncProfiler生成]
    C -->|No| F[何もしない]
    D -->|Yes| G{サンプリング判定}
    D -->|No| F
    G -->|選択| E
    G -->|非選択| F
    E --> H[profiler.start]
    H --> I[Async Profiler実行開始]
    I --> J[startWriting: DFS書き込みスレッド起動]
    J --> K[定期writeChunk]
    K --> L[stop → dump → resume]
    L --> M{最終チャンク?}
    M -->|No| K
    M -->|Yes| N[finishWriting]
    N --> O[inputStream/outputStream close]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-97-01 | サンプリング制御 | rand.nextInt(100)*0.01 < fractionでエグゼキューター選択 | executor.enabled=true |
| BR-97-02 | DFS書き込みサイクル | stop→dump→resume の3ステップでデータ一貫性確保 | dfsDir設定時 |
| BR-97-03 | ドライバーファイル名 | "profile-driver.jfr" | executorId = DRIVER_IDENTIFIER |
| BR-97-04 | エグゼキューターファイル名 | "profile-exec-{id}.jfr" | executorId != DRIVER_IDENTIFIER |
| BR-97-05 | ディレクトリ構造 | ${dfsDir}/${appId}[_${attemptId}]/${profileFile} | DFS出力時 |
| BR-97-06 | データ欠落許容 | stop→dump間のイベントは記録されない（コメントで言及） | DFS書き込み時 |

### 計算ロジック

サンプリング判定: `rand.nextInt(100) * 0.01 < executorProfilerFraction` -- 0.0-0.99の範囲の乱数とfraction値を比較

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

本機能はデータベースを直接操作しない。ファイルシステム（HDFS/ローカル）へのI/O操作のみ。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalArgumentException | ネイティブエラー | Async Profiler引数不正 | profilerOptions設定を確認 |
| IllegalStateException | ネイティブエラー | Async Profiler状態不正 | プロファイラの状態を確認 |
| IOException | I/Oエラー | DFS書き込み失敗 | ログ出力して継続（一部データ欠落） |
| IllegalArgumentException | 設定エラー | dfsDirがディレクトリでない | 有効なディレクトリパスを指定 |

### リトライ仕様

- writeChunk内のエラーはログ出力して処理を継続（プロファイリング自体は停止しない）
- finishWritingのInterruptedExceptionはスレッド割り込みを保持して終了

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

DFS書き込みはアトミックではない。stop→dump→resume サイクル間のイベントは記録されない。

## パフォーマンス要件

- DFS書き込みバッファ: UPLOAD_SIZE = 8MB
- DFS書き込み間隔: dfsWriteInterval（デフォルト30秒）
- Async Profilerデフォルト: wall clock 10ms間隔、alloc 2MB閾値、lock 10ms閾値

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

- DFS出力ディレクトリ: 770パーミッション
- DFS出力ファイル: 660パーミッション
- Hadoop設定を使用してDFSアクセス

## 備考

`connector/profiler/` パッケージ配下に実装。Spark 4.0で導入。Async Profiler（one.profiler）のネイティブライブラリに依存。全設定はversion 4.0.0から利用可能。

---

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

### 推奨読解順序

#### Step 1: 設定パラメータを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | package.scala | `connector/profiler/src/main/scala/org/apache/spark/profiler/package.scala` | **23-78行目**: 全設定パラメータの定義。ConfigBuilderによるデフォルト値・バリデーション |

**主要設定**:
- **25-30行目**: PROFILER_DRIVER_ENABLED（デフォルト: false）
- **32-37行目**: PROFILER_EXECUTOR_ENABLED（デフォルト: false）
- **39-46行目**: PROFILER_EXECUTOR_FRACTION（デフォルト: 0.1, [0.0, 1.0]）
- **48-53行目**: PROFILER_DFS_DIR（Optional）
- **55-61行目**: PROFILER_LOCAL_DIR（デフォルト: "."）
- **63-68行目**: PROFILER_ASYNC_PROFILER_OPTIONS（デフォルト: "event=wall,interval=10ms,alloc=2m,lock=10ms,chunktime=300s"）
- **70-76行目**: PROFILER_DFS_WRITE_INTERVAL（デフォルト: 30秒）

#### Step 2: プラグイン構造を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ProfilerPlugin.scala | `connector/profiler/src/main/scala/org/apache/spark/profiler/ProfilerPlugin.scala` | プラグインエントリーポイント |

**主要処理フロー**:
- **32-36行目**: ProfilerPluginクラス。SparkPluginを実装、driverPlugin/executorPluginを返却
- **38-64行目**: ProfilerDriverPlugin。init()でドライバープロファイリング開始、shutdown()で停止
- **45-53行目**: init()でPROFILER_DRIVER_ENABLED確認→SparkAsyncProfiler生成→start()
- **66-97行目**: ProfilerExecutorPlugin。init()でfraction判定→SparkAsyncProfiler生成→start()
- **81行目**: サンプリング判定: `rand.nextInt(100) * 0.01 < executorProfilerFraction`

#### Step 3: プロファイラ実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SparkAsyncProfiler.scala | `connector/profiler/src/main/scala/org/apache/spark/profiler/SparkAsyncProfiler.scala` | プロファイラのコア実装 |

**主要処理フロー**:
- **36-67行目**: フィールド定義。profilerOptions, dfsDirOpt, localDir, writeInterval, コマンド文字列、パーミッション定数
- **48-52行目**: プロファイルファイル名の決定（driver/exec分岐）
- **54-57行目**: 4つのプロファイラコマンド: startcmd, stopcmd, dumpcmd, resumecmd
- **59-60行目**: パーミッション定数: PROFILER_FOLDER_PERMISSIONS(770), PROFILER_FILE_PERMISSIONS(660)
- **71-78行目**: AsyncProfilerのロード。isSupported確認→extractionDir設定→load()
- **80-95行目**: start()。profiler.execute(startcmd) + startWriting()
- **98-107行目**: stop()。profiler.execute(stopcmd) + finishWriting()
- **116-144行目**: startWriting()。BufferedInputStream + ScheduledExecutorService + writeChunk定期実行
- **146-197行目**: writeChunk()。appId待機→DFS出力先決定→stop→dump→resume→read→write
- **199-217行目**: finishWriting()。threadpool.shutdown→awaitTermination→writeChunk(true)→close

**読解のコツ**: writeChunk()の処理フロー（146-197行目）がデータ転送の核心。stop→dump→resumeのサイクル中はプロファイリングイベントが記録されない点に注意（176-178行目のコメント参照）。

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

```
ProfilerPlugin
    |
    +-- ProfilerDriverPlugin
    |     +-- init(SparkContext, PluginContext)
    |     |     +-- SparkAsyncProfiler(conf, "driver")
    |     |           +-- AsyncProfilerLoader.load()
    |     |           +-- start()
    |     |                 +-- profiler.execute(startcmd)
    |     |                 +-- startWriting()
    |     |                       +-- ScheduledExecutorService
    |     |                             +-- writeChunk(false) [定期]
    |     +-- shutdown()
    |           +-- SparkAsyncProfiler.stop()
    |                 +-- profiler.execute(stopcmd)
    |                 +-- finishWriting()
    |
    +-- ProfilerExecutorPlugin
          +-- init(PluginContext, extraConf)
          |     +-- fraction判定
          |     +-- SparkAsyncProfiler(conf, executorId)
          |           +-- [同上]
          +-- shutdown()
                +-- [同上]
```

### データフロー図

```
[Async Profiler]          [SparkAsyncProfiler]           [DFS]

JVMイベント収集   ──>    ローカルJFRファイル
                          (localDir/profile-*.jfr)
                               |
                          writeChunk (定期)
                          stop→dump→resume
                               |
                          BufferedInputStream
                          (8MBバッファ読み込み)
                               |
                          FSDataOutputStream    ──>    dfsDir/appId/profile-*.jfr
                          (書き込み)                    (770/660パーミッション)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ProfilerPlugin.scala | `connector/profiler/src/main/scala/org/apache/spark/profiler/ProfilerPlugin.scala` | ソース | Sparkプラグインエントリーポイント |
| SparkAsyncProfiler.scala | `connector/profiler/src/main/scala/org/apache/spark/profiler/SparkAsyncProfiler.scala` | ソース | Async Profilerラッパー・DFS書き込み |
| package.scala | `connector/profiler/src/main/scala/org/apache/spark/profiler/package.scala` | ソース | 設定パラメータ定義 |
