# 機能設計書 6-ストリーム実行環境

## 概要

本ドキュメントは、Apache FlinkのStreamExecutionEnvironment（ストリーム実行環境）について詳細に記載する。StreamExecutionEnvironmentはストリーム処理ジョブの設定と実行のためのコンテキストを提供し、DataStream APIの起点となる重要なクラスである。

### 本機能の処理概要

StreamExecutionEnvironmentは、ストリーミングプログラムが実行されるコンテキストを提供する。LocalStreamEnvironmentは現在のJVM内での実行を、RemoteStreamEnvironmentはリモートセットアップでの実行を行う。このクラスはジョブ実行の制御（並列度やフォールトトレランス/チェックポイントパラメータの設定）および外部とのインタラクション（データアクセス）のためのメソッドを提供する。

**業務上の目的・背景**：ストリーム処理ジョブを実行するには、実行環境の設定、データソースの定義、変換処理の記述、シンクへの出力という一連のフローが必要である。StreamExecutionEnvironmentはこのフローのエントリーポイントとして、並列度設定、チェックポイント設定、時間特性（イベント時間/処理時間）の設定など、ジョブ全体に影響する設定を集約管理する。これにより、開発者は個々のオペレーターの詳細を意識することなく、ジョブ全体の特性を統一的に制御できる。

**機能の利用シーン**：
- ストリーム処理ジョブの初期化
- DataStreamSourceの生成
- ジョブのサブミットと実行
- チェックポイント/セーブポイントの設定

**主要な処理内容**：
1. getExecutionEnvironment()：実行環境の取得（ローカル/リモート自動判定）
2. fromSource()：Sourceコネクタからのストリーム生成
3. fromElements()：コレクションからのストリーム生成
4. execute()：ジョブの実行
5. enableCheckpointing()：チェックポイントの有効化

**関連システム・外部連携**：PipelineExecutorを通じてJobManagerと連携し、ジョブをサブミットする。また、各種ソースコネクタ（Kafka、ファイル等）との統合ポイントを提供する。

**権限による制御**：ジョブサブミット時にクラスタへのアクセス権限が必要。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 2 | ジョブサブミット | 主機能 | ジョブ実行のREST API呼び出し |
| 3 | 実行中ジョブ一覧 | 結果表示画面 | execute()で開始されたジョブの表示 |

## 機能種別

ジョブ管理 / 実行環境

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| parallelism | int | No | デフォルト並列度 | > 0 |
| checkpointInterval | long | No | チェックポイント間隔（ms） | > 0 |
| configuration | Configuration | No | 追加設定 | - |

### 入力データソース

- プログラムからの動的設定
- flink-conf.yaml設定ファイル

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| JobExecutionResult | JobExecutionResult | ジョブ実行結果 |
| DataStreamSource | DataStreamSource<T> | データストリームソース |

### 出力先

- JobManager（ジョブサブミット）
- コンソール（execute結果）

## 処理フロー

### 処理シーケンス

```
1. 実行環境取得
   └─ getExecutionEnvironment()でコンテキストに応じた環境を取得
2. ソース定義
   └─ fromSource()/fromElements()等でDataStreamSourceを生成
3. 変換処理定義
   └─ map/filter/keyBy等の変換オペレーター適用
4. シンク定義
   └─ sinkTo()で出力先を指定
5. ジョブ実行
   └─ execute()でStreamGraphを生成しJobManagerにサブミット
```

### フローチャート

```mermaid
flowchart TD
    A[getExecutionEnvironment] --> B[ソース定義]
    B --> C[変換処理定義]
    C --> D[シンク定義]
    D --> E[execute]
    E --> F[StreamGraph生成]
    F --> G[JobGraph生成]
    G --> H[JobManagerサブミット]
    H --> I[JobExecutionResult]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | 環境自動判定 | CLI経由ならリモート、直接実行ならローカル | getExecutionEnvironment()時 |
| BR-002 | 並列度デフォルト | 未設定時は利用可能CPUコア数 | ローカル実行時 |
| BR-003 | Exactly-once | チェックポイント有効時にExactly-once保証 | enableCheckpointing()時 |

### 計算ロジック

特になし。

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

本機能は直接データベース操作を行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | InvalidProgramException | 無効なプログラム構成 | プログラムを修正 |
| - | JobExecutionException | ジョブ実行失敗 | ログを確認し原因を特定 |

### リトライ仕様

チェックポイントが有効な場合、障害時は最後のチェックポイントから自動リカバリ。

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

チェックポイントとTwo-Phase Commitを組み合わせてExactly-once保証を実現。

## パフォーマンス要件

- execute()呼び出しからジョブ開始までは数秒以内
- StreamGraph/JobGraph生成はO(オペレーター数)

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

- リモート実行時はクラスタへの認証が必要
- JAR/クラスパスの管理に注意

## 備考

StreamExecutionEnvironmentは@Publicアノテーションが付与された安定したAPIである。

---

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

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

### 推奨読解順序

#### Step 1: 基本クラスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StreamExecutionEnvironment.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java` | ストリーム実行環境のメインクラス |

**読解のコツ**:
- **129-142行目**: JavaDocでクラスの役割を理解
- **142行目**: @Publicアノテーションとクラス定義
- **166行目**: ExecutionConfig保持
- **169行目**: CheckpointConfig保持
- **171行目**: transformationsリスト
- **194行目**: Configuration保持
- getExecutionEnvironment()静的メソッド
- fromSource()、fromElements()メソッド
- execute()メソッド

#### Step 2: ジョブ実行フローを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | StreamGraphGenerator.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java` | StreamGraph生成 |
| 2-2 | StreamGraph.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java` | ストリームグラフ表現 |

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

```
StreamExecutionEnvironment.getExecutionEnvironment()
    │
    └─ fromSource() / fromElements()
           │
           └─ DataStreamSource<T>
                  │
                  └─ transformations追加
                         │
                         └─ execute()
                                │
                                ├─ StreamGraphGenerator
                                │      └─ StreamGraph生成
                                │
                                └─ PipelineExecutor
                                       └─ JobGraph生成
                                              └─ JobManager送信
```

### データフロー図

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

設定/ソース定義 ───▶ StreamExecutionEnvironment ───▶ DataStream
     │                      │                           │
     │                      ▼                           │
     │               StreamGraph生成                    │
     │                      │                           │
     │                      ▼                           │
     └────────────▶   execute()   ─────────────▶ JobExecutionResult
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StreamExecutionEnvironment.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java` | ソース | 実行環境メインクラス |
| LocalStreamEnvironment.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java` | ソース | ローカル実行環境 |
| RemoteStreamEnvironment.java | `flink-clients/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java` | ソース | リモート実行環境 |
| ExecutionConfig.java | `flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java` | ソース | 実行設定 |
| CheckpointConfig.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java` | ソース | チェックポイント設定 |
| StreamGraph.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java` | ソース | ストリームグラフ |
| DataStreamSource.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSource.java` | ソース | データストリームソース |
