# 機能設計書 67-YARN

## 概要

本ドキュメントは、Apache FlinkのYARNデプロイメント機能（flink-yarn モジュール）の設計仕様を記載する。

### 本機能の処理概要

YARN（Yet Another Resource Negotiator）デプロイメント機能は、FlinkクラスターをApache Hadoop YARNクラスター上にデプロイするための機能である。YARNのリソース管理機能を活用して、FlinkのJobManagerとTaskManagerを動的にプロビジョニングする。

**業務上の目的・背景**：YARNはHadoopエコシステムの標準的なリソースマネージャーであり、多くの企業がHadoopクラスターを運用している。本機能によりFlinkを既存のHadoopインフラ上で実行でき、リソースの共有と効率的な利用が可能になる。

**機能の利用シーン**：
- Hadoopクラスター上でのFlink実行
- YARNのリソース管理機能を活用した動的スケーリング
- セッションクラスターとアプリケーションクラスターの両モードでのデプロイ
- Kerberosセキュリティ環境でのFlink実行

**主要な処理内容**：
1. YARNクライアントを使用したアプリケーション登録
2. JobManager（ApplicationMaster）のコンテナ起動
3. TaskManagerコンテナの動的割り当て
4. リソースファイルのHDFSへのアップロード
5. Kerberos認証とデリゲーショントークン管理

**関連システム・外部連携**：
- Apache Hadoop YARN ResourceManager
- HDFS（リソースファイルの格納）
- Kerberos KDC（セキュリティ環境の場合）

**権限による制御**：YARNキュー権限、Kerberos認証

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は画面を持たないインフラ機能 |

## 機能種別

クラスターデプロイメント（YARNリソースマネージャー連携）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| yarn.application.queue | String | No | YARNキュー名 | - |
| yarn.application.name | String | No | アプリケーション名 | - |
| yarn.application.type | String | No | アプリケーションタイプ | - |
| yarn.application.priority | Integer | No | 優先度（-1でデフォルト） | 非負整数または-1 |
| yarn.appmaster.vcores | Integer | No | AM用vcore数（デフォルト: 1） | 正整数 |
| yarn.containers.vcores | Integer | No | コンテナ用vcore数 | -1でスロット数に連動 |
| yarn.application-attempts | Integer | No | AM再起動回数 | - |
| yarn.heartbeat.interval | Integer | No | ハートビート間隔（秒、デフォルト: 5） | 正整数 |
| yarn.ship-files | List<String> | No | 転送ファイルリスト | セミコロン区切り |
| yarn.ship-archives | List<String> | No | 転送アーカイブリスト | セミコロン区切り |
| yarn.provided.lib.dirs | List<String> | No | 事前配置ライブラリ | セミコロン区切り |
| yarn.application.node-label | String | No | ノードラベル | - |
| yarn.flink-dist-jar | String | No | flink-distのパス | - |
| yarn.staging-directory | String | No | ステージングディレクトリ | - |

### 入力データソース

- Flink設定ファイル（flink-conf.yaml）
- Hadoop設定ファイル（core-site.xml、hdfs-site.xml、yarn-site.xml）
- JobGraph（ジョブデプロイ時）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| ApplicationId | String | YARNアプリケーションID |
| TrackingURL | String | ApplicationMasterのトラッキングURL |
| ApplicationReport | Object | アプリケーション状態レポート |

### 出力先

- YARN ResourceManager（アプリケーション登録）
- HDFS（リソースファイル）
- クライアントコンソール（ログ出力）

## 処理フロー

### 処理シーケンス

```
1. デプロイメント準備（isReadyForDeployment）
   └─ flinkJarPath確認
   └─ vcore数のYARN上限チェック
   └─ Hadoop環境変数確認

2. YARNキュー確認（checkYarnQueues）
   └─ 指定キューの存在確認
   └─ キューリソースの取得

3. アプリケーション作成
   └─ yarnClient.createApplication()
   └─ 最大リソースキャパシティ取得
   └─ クラスターリソース検証

4. ApplicationMaster起動（startAppMaster）
   └─ リソースファイルのHDFSアップロード
   └─ ContainerLaunchContext設定
   └─ ApplicationSubmissionContext設定
   └─ yarnClient.submitApplication()

5. アプリケーション起動待機
   └─ ApplicationReportのポーリング
   └─ RUNNING状態まで待機

6. TaskManager起動（ResourceManagerDriver）
   └─ AMRMClient.ContainerRequest
   └─ コンテナ割り当てコールバック
   └─ NMClient経由でコンテナ起動
```

### フローチャート

```mermaid
flowchart TD
    A[開始: deploySessionCluster/deployApplicationCluster] --> B[isReadyForDeployment]
    B --> C{準備OK?}
    C -->|No| D[YarnDeploymentException]
    C -->|Yes| E[checkYarnQueues]
    E --> F[yarnClient.createApplication]
    F --> G[validateClusterResources]
    G --> H{リソース足りる?}
    H -->|No| D
    H -->|Yes| I[startAppMaster]
    I --> J[HDFSへファイルアップロード]
    J --> K[ContainerLaunchContext作成]
    K --> L[ApplicationSubmissionContext作成]
    L --> M[yarnClient.submitApplication]
    M --> N[ApplicationReport待機]
    N --> O{RUNNING?}
    O -->|No| P[タイムアウト/エラー]
    O -->|Yes| Q[RestClusterClient返却]
    Q --> Z[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | vcore上限チェック | 設定vcoreがYARN最大vcoreを超えてはならない | デプロイ時 |
| BR-02 | メモリ上限チェック | JM/TMメモリがYARN最大メモリを超えてはならない | デプロイ時 |
| BR-03 | 最小割り当てメモリ | YARNの最小割り当て単位以上である必要 | デプロイ時 |
| BR-04 | Kerberos資格情報 | Kerberos有効時は有効な資格情報が必要 | デプロイ時 |
| BR-05 | 環境変数チェック | HADOOP_CONF_DIRまたはYARN_CONF_DIRが必要 | 警告のみ |
| BR-06 | AM再起動制限 | HA無効時1回、HA有効時2回がデフォルト | - |

### 計算ロジック

メモリ正規化:
```
normalizedMemMB = ((componentMemoryMB + yarnMinAllocationMB - 1) / yarnMinAllocationMB) * yarnMinAllocationMB
```

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

本機能はデータベースを使用しない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| YarnDeploymentException | デプロイエラー | flinkJarPath未設定 | flink-distパス設定 |
| IllegalConfigurationException | 設定エラー | vcore超過 | vcore設定修正 |
| YarnDeploymentException | リソースエラー | メモリ不足 | メモリ設定修正 |
| RuntimeException | 認証エラー | Kerberos認証失敗 | 認証情報確認 |
| ClusterRetrieveException | 取得エラー | アプリケーション取得失敗 | ApplicationId確認 |

### リトライ仕様

- ApplicationMasterはyarn.application-attemptsで指定された回数まで再起動
- コンテナ割り当ては継続的にリクエスト

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

- HDFSへのファイルアップロードは個別ファイル単位
- アプリケーション登録はYARN側でアトミック

## パフォーマンス要件

- ハートビート間隔はyarn.heartbeat.intervalで設定（デフォルト5秒）
- コンテナリクエスト時は500msの高頻度ハートビート
- 並列ファイルアップロードによる起動時間短縮

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

- Kerberos認証のサポート
- デリゲーショントークンの取得と配布
- keytabファイルのローカライズ
- YARNのACL（VIEW/MODIFY）設定対応

## 備考

- Session ClusterとApplication Clusterの2つのデプロイモードをサポート
- YarnClusterDescriptorがClusterDescriptorインターフェースを実装
- YarnResourceManagerDriverがリソース管理を担当

---

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

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

### 推奨読解順序

#### Step 1: 設定オプションを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | YarnConfigOptions.java | `flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java` | YARN固有の設定項目定義 |
| 1-2 | YarnDeploymentTarget.java | `flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnDeploymentTarget.java` | デプロイモード（Session/Application） |

**読解のコツ**: ConfigOptionで定義される各設定項目の意味とデフォルト値を確認。特にAPP_MASTER_VCORES、CONTAINERS_VCORES、HEARTBEAT_DELAY_SECONDS等が重要。

#### Step 2: クラスターディスクリプタを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | YarnClusterDescriptor.java | `flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java` | メインデプロイメントクラス |

**主要処理フロー（YarnClusterDescriptor.java）**:
- **153-199行目**: フィールド定義（YarnClient、YarnConfiguration等）
- **200-233行目**: コンストラクタ（設定読み込み、ファイル転送リスト構築）
- **389-431行目**: isReadyForDeployment()でデプロイ前検証
- **442-446行目**: close()でYarnClient停止
- **452-490行目**: retrieve()で既存アプリケーション取得
- **493-505行目**: deploySessionCluster()でセッションクラスター起動
- **508-550行目**: deployApplicationCluster()でアプリケーションクラスター起動
- **581-698行目**: deployInternal()でメインデプロイメント処理
- **700-796行目**: validateClusterResources()でリソース検証

#### Step 3: リソースマネージャードライバーを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | YarnResourceManagerDriver.java | `flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManagerDriver.java` | TaskManager管理 |

**主要処理フロー（YarnResourceManagerDriver.java）**:
- **83-131行目**: クラス定義とフィールド（AMRMClientAsync、NMClientAsync等）
- **133-176行目**: コンストラクタ（ハートビート間隔設定等）
- **183-200行目**: initializeInternal()でResourceManagerClient初期化

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | YarnSessionClusterEntrypoint.java | `flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnSessionClusterEntrypoint.java` | セッションモードのAMエントリーポイント |
| 4-2 | YarnApplicationClusterEntryPoint.java | `flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java` | アプリケーションモードのAMエントリーポイント |

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

```
YarnClusterDescriptor
    │
    ├─ deploySessionCluster() / deployApplicationCluster()
    │      │
    │      └─ deployInternal()
    │             │
    │             ├─ isReadyForDeployment()
    │             │      └─ vcore/メモリ検証
    │             │
    │             ├─ checkYarnQueues()
    │             │      └─ YarnClient.getQueueInfo()
    │             │
    │             ├─ yarnClient.createApplication()
    │             │      └─ GetNewApplicationResponse
    │             │
    │             ├─ validateClusterResources()
    │             │      └─ メモリ/リソース検証
    │             │
    │             └─ startAppMaster()
    │                    │
    │                    ├─ YarnApplicationFileUploader
    │                    │      └─ HDFSへファイルアップロード
    │                    │
    │                    ├─ ContainerLaunchContext作成
    │                    │
    │                    └─ yarnClient.submitApplication()
    │
    └─ retrieve()
           └─ yarnClient.getApplicationReport()

YarnResourceManagerDriver（ApplicationMaster内）
    │
    ├─ AMRMClientAsync
    │      ├─ registerApplicationMaster()
    │      ├─ addContainerRequest()
    │      └─ onContainersAllocated()
    │
    └─ NMClientAsync
           └─ startContainerAsync()
```

### データフロー図

```
[クライアント側]                                          [YARN側]

flink-conf.yaml ─────┐
                     │
yarn-site.xml ───────┼──▶ YarnClusterDescriptor
                     │           │
hadoop config ───────┘           │
                                 ▼
                    ┌────────────────────────┐
                    │  YarnClient            │
                    │  - createApplication() │
                    │  - submitApplication() │
                    └───────────┬────────────┘
                                │
                    ┌───────────▼────────────┐
                    │  HDFS                  │
                    │  - flink-dist.jar      │──────────▶  ResourceManager
                    │  - user libs           │                    │
                    │  - flink-conf.yaml     │                    ▼
                    └────────────────────────┘           ┌────────────────┐
                                                         │ ApplicationMaster│
                                                         │ (JobManager)   │
                                                         └───────┬────────┘
                                                                 │
                                              ┌──────────────────┼──────────────────┐
                                              ▼                  ▼                  ▼
                                        ┌──────────┐      ┌──────────┐      ┌──────────┐
                                        │Container │      │Container │      │Container │
                                        │   (TM)   │      │   (TM)   │      │   (TM)   │
                                        └──────────┘      └──────────┘      └──────────┘
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| YarnClusterDescriptor.java | `flink-yarn/src/main/java/org/apache/flink/yarn/` | ソース | メインデプロイメントクラス |
| YarnResourceManagerDriver.java | `flink-yarn/src/main/java/org/apache/flink/yarn/` | ソース | TaskManager管理 |
| YarnConfigOptions.java | `flink-yarn/src/main/java/org/apache/flink/yarn/configuration/` | ソース | YARN設定オプション |
| YarnDeploymentTarget.java | `flink-yarn/src/main/java/org/apache/flink/yarn/configuration/` | ソース | デプロイモード定義 |
| YarnSessionClusterEntrypoint.java | `flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/` | ソース | セッションモードAMエントリーポイント |
| YarnApplicationClusterEntryPoint.java | `flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/` | ソース | アプリケーションモードAMエントリーポイント |
| YarnApplicationFileUploader.java | `flink-yarn/src/main/java/org/apache/flink/yarn/` | ソース | HDFSファイルアップロード |
| Utils.java | `flink-yarn/src/main/java/org/apache/flink/yarn/` | ソース | ユーティリティ関数 |
| YarnClusterClientFactory.java | `flink-yarn/src/main/java/org/apache/flink/yarn/` | ソース | クライアントファクトリ |
