# 機能設計書 82-ストリーム例

## 概要

本ドキュメントは、Apache Flinkのストリーム処理サンプルプログラム（flink-examples-streaming）の機能設計書である。開発者がFlink DataStream APIの使用方法を学ぶためのリファレンス実装を提供する。

### 本機能の処理概要

flink-examples-streamingモジュールは、Apache Flink DataStream APIの使用方法を示すサンプルプログラム集である。WordCount、ウィンドウ処理、非同期I/O、状態機械など、典型的なストリーム処理パターンの実装例を提供する。

**業務上の目的・背景**：Flinkを初めて使用する開発者にとって、公式のサンプルコードは最も重要な学習リソースである。実際に動作するコードを参照することで、APIの使用方法、ベストプラクティス、典型的な処理パターンを効率的に学習できる。また、新機能のデモンストレーションや、パフォーマンステスト用のベースラインとしても活用される。

**機能の利用シーン**：(1)Flink学習時のリファレンスコード、(2)新規プロジェクト開始時のテンプレート、(3)APIドキュメントの補完資料、(4)機能検証・デモンストレーション、(5)パフォーマンスベンチマーク

**主要な処理内容**：
1. **WordCount**: テキストデータから単語の出現頻度をカウント
2. **WindowWordCount**: ウィンドウを使用した単語カウント
3. **SessionWindowing**: セッションウィンドウの使用例
4. **TopSpeedWindowing**: 複合的なウィンドウ処理の例
5. **WindowJoin**: 複数ストリームのウィンドウ結合
6. **AsyncIOExample**: 非同期I/O操作の実装例
7. **StateMachineExample**: 状態機械パターンの実装例
8. **SideOutputExample**: サイドアウトプットの使用例

**関連システム・外部連携**：ファイルシステム（入出力）、ソケット（入力）、Kafka（StateMachineExample）

**権限による制御**：なし（サンプルプログラム）

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | サンプルプログラムのため関連画面なし |

## 機能種別

サンプルコード / 学習リソース / リファレンス実装

## 入力仕様

### 入力パラメータ（WordCount例）

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| --input | Path | No | 入力ファイル/ディレクトリパス | 存在するパスであること |
| --output | Path | No | 出力ディレクトリパス | 書き込み可能であること |
| --discovery-interval | Duration | No | 新規ファイル監視間隔 | 正の期間であること |
| --execution-mode | Enum | No | 実行モード（BATCH/STREAMING/AUTOMATIC） | 有効な値であること |
| --async-state | Boolean | No | 非同期ステートの有効化 | - |

### 入力データソース

- テキストファイル（FileSource経由）
- インメモリデータ（WordCountData.WORDS）
- ソケットストリーム（SocketWindowWordCount）
- Kafkaトピック（StateMachineExample）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| word | String | 単語 |
| count | Integer | 出現回数 |

### 出力先

- ファイルシステム（FileSink）
- 標準出力（print()）
- コンソール（開発/デバッグ用）

## 処理フロー

### 処理シーケンス（WordCount）

```
1. StreamExecutionEnvironment取得
   └─ getExecutionEnvironment()
2. 実行モード設定
   └─ env.setRuntimeMode(params.getExecutionMode())
3. 入力ソース設定
   └─ FileSource or fromData(WordCountData.WORDS)
4. 変換処理
   ├─ flatMap(Tokenizer) - 単語分割
   └─ keyBy(word) - キーによるグループ化
5. 集計処理
   └─ sum(1) - カウント合計
6. 出力処理
   └─ FileSink or print()
7. ジョブ実行
   └─ env.execute("WordCount")
```

### フローチャート

```mermaid
flowchart TD
    A[main開始] --> B[StreamExecutionEnvironment取得]
    B --> C[実行モード設定]
    C --> D{入力パス指定?}
    D -->|Yes| E[FileSource作成]
    D -->|No| F[インメモリデータ使用]
    E --> G[DataStream作成]
    F --> G
    G --> H[flatMap - Tokenizer]
    H --> I[keyBy - 単語でグループ化]
    I --> J[sum - カウント集計]
    J --> K{出力パス指定?}
    K -->|Yes| L[FileSinkへ出力]
    K -->|No| M[print - 標準出力]
    L --> N[execute実行]
    M --> N
    N --> O[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | 単語正規化 | 入力テキストを小文字に変換し、英数字以外で分割 | Tokenizer処理時 |
| BR-002 | 空単語除外 | 長さ0の単語はスキップ | Tokenizer処理時 |
| BR-003 | 自動実行モード | 入力が有界の場合BATCH、無界の場合STREAMING | AUTOMATIC指定時 |

### 計算ロジック

- 単語カウント：各単語を(word, 1)のタプルに変換し、keyBy後にsum(1)で集計

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

### 操作別データベース影響一覧

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | FileNotFoundException | 入力ファイルが見つからない | パスを確認 |
| - | IOException | ファイル読み書きエラー | アクセス権限を確認 |

### リトライ仕様

サンプルプログラムのため特別なリトライ機構は実装されていない。

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

なし（ストリーム処理のサンプル）

## パフォーマンス要件

サンプルプログラムのため特定のパフォーマンス要件はない。

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

サンプルプログラムのため特別なセキュリティ対策は含まれていない。

## 備考

- 各サンプルプログラムは独立して実行可能
- flink runコマンドでJARファイルとして実行可能
- DSv2（DataStream V2）APIのサンプルも含まれている

---

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

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

### 推奨読解順序

#### Step 1: 基本的なWordCountを理解する

最も基本的なDataStream APIの使用パターンを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | WordCount.java | `flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java` | DataStream APIの基本パターン |

**主要処理フロー**:
- **78-92行目**: StreamExecutionEnvironment取得と非同期ステート設定
- **113-134行目**: データソースの設定（FileSource or fromData）
- **136-157行目**: flatMap→keyBy→sumの変換パイプライン
- **159-175行目**: 出力先の設定（FileSink or print）
- **179行目**: env.execute()でジョブ実行

**読解のコツ**: Flinkの基本パターンは「Environment取得」→「Source設定」→「変換処理」→「Sink設定」→「実行」である。

#### Step 2: ユーザー定義関数を理解する

Tokenizerクラスで、FlatMapFunctionの実装方法を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | WordCount.java | `flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java` | Tokenizerクラス（191-206行目） |

**主要処理フロー**:
- **195行目**: flatMapメソッドの実装
- **197行目**: 小文字変換と単語分割
- **200-203行目**: Collectorへの出力

#### Step 3: ウィンドウ処理を理解する

ウィンドウを使用したストリーム処理パターンを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TopSpeedWindowing.java | `flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java` | 複合ウィンドウ処理 |
| 3-2 | SessionWindowing.java | `flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java` | セッションウィンドウ |

#### Step 4: 非同期I/Oを理解する

外部システムとの非同期通信パターンを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | AsyncIOExample.java | `flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/async/AsyncIOExample.java` | 非同期I/Oの実装 |

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

```
WordCount.main()
    │
    ├─ CLI.fromArgs() - コマンドライン引数解析
    │
    ├─ StreamExecutionEnvironment.getExecutionEnvironment()
    │
    ├─ env.setRuntimeMode() - 実行モード設定
    │
    ├─ FileSource.forRecordStreamFormat() or env.fromData()
    │      └─ TextLineInputFormat
    │
    ├─ text.flatMap(Tokenizer)
    │      └─ Tokenizer.flatMap()
    │
    ├─ .keyBy(value -> value.f0)
    │
    ├─ .sum(1)
    │
    ├─ FileSink.forRowFormat() or .print()
    │
    └─ env.execute("WordCount")
```

### データフロー図

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

テキストファイル     flatMap(Tokenizer)
     or           ┌────────────────────┐
インメモリデータ ──▶│ "hello world" ──▶ │──▶ (hello, 1)
                  │   split & emit     │    (world, 1)
                  └────────────────────┘
                           │
                    keyBy(word)
                  ┌────────────────────┐
                  │ グループ化          │
                  │ hello: [(hello,1)] │
                  │ world: [(world,1)] │
                  └────────────────────┘
                           │
                       sum(1)
                  ┌────────────────────┐
                  │ カウント集計        │──▶ (hello, 5)   ──▶ ファイル
                  │ hello: 1+1+1+1+1   │    (world, 3)       or
                  └────────────────────┘                    標準出力
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| WordCount.java | `flink-examples/flink-examples-streaming/src/main/java/.../wordcount/WordCount.java` | ソース | 基本的なWordCountサンプル |
| CLI.java | `flink-examples/flink-examples-streaming/src/main/java/.../wordcount/util/CLI.java` | ソース | コマンドライン引数ユーティリティ |
| WordCountData.java | `flink-examples/flink-examples-streaming/src/main/java/.../wordcount/util/WordCountData.java` | ソース | サンプルデータ定義 |
| WindowWordCount.java | `flink-examples/flink-examples-streaming/src/main/java/.../windowing/WindowWordCount.java` | ソース | ウィンドウ版WordCount |
| TopSpeedWindowing.java | `flink-examples/flink-examples-streaming/src/main/java/.../windowing/TopSpeedWindowing.java` | ソース | 複合ウィンドウ処理サンプル |
| SessionWindowing.java | `flink-examples/flink-examples-streaming/src/main/java/.../windowing/SessionWindowing.java` | ソース | セッションウィンドウサンプル |
| WindowJoin.java | `flink-examples/flink-examples-streaming/src/main/java/.../join/WindowJoin.java` | ソース | ウィンドウ結合サンプル |
| AsyncIOExample.java | `flink-examples/flink-examples-streaming/src/main/java/.../async/AsyncIOExample.java` | ソース | 非同期I/Oサンプル |
| StateMachineExample.java | `flink-examples/flink-examples-streaming/src/main/java/.../statemachine/StateMachineExample.java` | ソース | 状態機械サンプル |
| SideOutputExample.java | `flink-examples/flink-examples-streaming/src/main/java/.../sideoutput/SideOutputExample.java` | ソース | サイドアウトプットサンプル |
| SocketWindowWordCount.java | `flink-examples/flink-examples-streaming/src/main/java/.../socket/SocketWindowWordCount.java` | ソース | ソケット入力サンプル |
