# 機能設計書 8-ProcessFunction

## 概要

本ドキュメントは、Apache FlinkのProcessFunction機能について詳細に記載する。ProcessFunctionは低レベルストリーム処理関数として、タイマー、ステート、サイドアウトプットなど高度な機能へのアクセスを提供する。

### 本機能の処理概要

ProcessFunctionは、すべてのユーザー定義プロセス関数の基底インターフェースである。基本的なmap/filter等の変換では実現できない高度な処理（タイマーベースの処理、複数出力、ステート管理等）を実装するために使用される。OneInputStreamProcessFunctionは単一入力ストリームの処理を、TwoInputStreamProcessFunctionは2つの入力ストリームの処理を担当する。

**業務上の目的・背景**：ストリーム処理において、単純なデータ変換だけでなく、時間ベースの処理（タイムアウト検知、セッション管理等）や複雑なステートフルな処理が必要となることがある。ProcessFunctionは、イベント時間/処理時間タイマーの登録、キー付きステートへのアクセス、サイドアウトプットによる複数ストリームへの出力など、Flinkの低レベル機能への直接アクセスを提供する。これにより、開発者はビジネスロジックに応じた柔軟なストリーム処理を実装できる。

**機能の利用シーン**：
- タイムアウト検知（一定時間イベントがない場合のアラート）
- セッション管理（ユーザーセッションの追跡）
- 複雑なイベントパターン検出
- 複数出力先への分岐処理

**主要な処理内容**：
1. processRecord()：各レコードの処理
2. onProcessingTimer()：処理時間タイマーコールバック
3. onWatermark()：ウォーターマーク受信時コールバック
4. usesStates()：使用するステートの宣言
5. declareWatermarks()：ウォーターマークの宣言

**関連システム・外部連携**：ProcessFunctionはTaskManagerで実行され、RuntimeContextを通じてステートバックエンド、タイマーサービスと連携する。

**権限による制御**：ProcessFunction自体には権限制御は存在しない。

## 関連画面

本機能はバックエンドのデータ処理機能であり、直接関連する画面は存在しない。

## 機能種別

データ変換 / ステートフル処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| record | IN | Yes | 処理対象のレコード | - |
| output | Collector<OUT> | Yes | 出力コレクター | Not null |
| ctx | PartitionedContext | Yes | 実行コンテキスト | Not null |

### 入力データソース

- 上流オペレーターからのストリームデータ
- タイマーからのコールバック

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| output | OUT | 処理結果 |
| WatermarkHandlingResult | Enum | ウォーターマーク処理結果 |

### 出力先

- 下流オペレーター
- サイドアウトプット（複数出力対応）

## 処理フロー

### 処理シーケンス

```
1. 関数初期化
   └─ open(NonPartitionedContext)で初期化処理
2. レコード処理
   └─ processRecord(record, output, ctx)で各レコードを処理
3. タイマー処理
   └─ onProcessingTimer(timestamp, output, ctx)でタイマーコールバック
4. ウォーターマーク処理
   └─ onWatermark(watermark, output, ctx)でウォーターマークを処理
5. 終了処理
   └─ endInput(ctx)で入力終了を通知
   └─ close()でリソース解放
```

### フローチャート

```mermaid
flowchart TD
    A[open] --> B{イベント種別}
    B -->|レコード| C[processRecord]
    B -->|タイマー| D[onProcessingTimer]
    B -->|ウォーターマーク| E[onWatermark]
    C --> F[出力]
    D --> F
    E --> G[WatermarkHandlingResult]
    F --> B
    B -->|入力終了| H[endInput]
    H --> I[close]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | ステート宣言 | 使用するステートはusesStates()で事前宣言が必要 | ステート使用時 |
| BR-002 | ウォーターマーク宣言 | ウォーターマークはdeclareWatermarks()で事前宣言 | ウォーターマーク使用時 |
| BR-003 | タイマー登録 | コンテキスト経由でタイマーを登録 | タイマー使用時 |

### 計算ロジック

特になし。ユーザー定義のビジネスロジックに依存。

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | Exception | processRecord()内で例外 | フェイルオーバー処理 |
| - | IllegalStateException | コンテキスト未初期化 | ライフサイクルを確認 |

### リトライ仕様

チェックポイントが有効な場合、障害時は最後のチェックポイントから再実行。

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

ProcessFunction自体にはトランザクション機能はない。チェックポイントと連携してExactly-once保証。

## パフォーマンス要件

- processRecord()は各レコードに対して呼び出されるため軽量実装が推奨
- ステートアクセスはキャッシュを活用して効率化

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

- ProcessFunctionはシリアライズされてTaskManagerに転送される

## 備考

ProcessFunctionは@Experimentalアノテーションが付与されており、APIが変更される可能性がある。

---

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

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

### 推奨読解順序

#### Step 1: 基本インターフェースを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ProcessFunction.java | `flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/ProcessFunction.java` | 基底インターフェース |

**読解のコツ**:
- **29行目**: JavaDocでProcessFunctionの役割を理解
- **31行目**: @ExperimentalアノテーションとFunctionインターフェース継承
- **38-40行目**: usesStates()のデフォルト実装
- **48-50行目**: declareWatermarks()のデフォルト実装
- **62行目**: close()のデフォルト実装

#### Step 2: 単一入力処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | OneInputStreamProcessFunction.java | `flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java` | 単一入力処理 |

**主要処理フロー**:
- **30行目**: ProcessFunctionを継承
- **41行目**: open()メソッド
- **50-51行目**: processRecord()抽象メソッド
- **59行目**: endInput()メソッド
- **68-69行目**: onProcessingTimer()メソッド
- **72-76行目**: onWatermark()メソッド

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

```
ProcessFunction
    │
    └─ OneInputStreamProcessFunction<IN, OUT>
           │
           ├─ open(NonPartitionedContext)
           │
           ├─ processRecord(IN, Collector<OUT>, PartitionedContext)
           │
           ├─ onProcessingTimer(timestamp, Collector, PartitionedContext)
           │
           ├─ onWatermark(Watermark, Collector, NonPartitionedContext)
           │
           ├─ endInput(NonPartitionedContext)
           │
           └─ close()
```

### データフロー図

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

レコード ───────▶ processRecord() ───────▶ Collector<OUT>
     │                   │
タイマー ───────▶ onProcessingTimer() ───▶ Collector<OUT>
     │                   │
ウォーターマーク ─▶ onWatermark() ─────▶ WatermarkHandlingResult
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ProcessFunction.java | `flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/ProcessFunction.java` | ソース | 基底インターフェース |
| OneInputStreamProcessFunction.java | `flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java` | ソース | 単一入力処理 |
| TwoInputNonBroadcastStreamProcessFunction.java | `flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputNonBroadcastStreamProcessFunction.java` | ソース | 2入力処理 |
| TwoOutputStreamProcessFunction.java | `flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputStreamProcessFunction.java` | ソース | 2出力処理 |
| PartitionedContext.java | `flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/PartitionedContext.java` | ソース | パーティションコンテキスト |
| NonPartitionedContext.java | `flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/NonPartitionedContext.java` | ソース | 非パーティションコンテキスト |
| StateDeclaration.java | `flink-core-api/src/main/java/org/apache/flink/api/common/state/StateDeclaration.java` | ソース | ステート宣言 |
| WatermarkDeclaration.java | `flink-core-api/src/main/java/org/apache/flink/api/common/watermark/WatermarkDeclaration.java` | ソース | ウォーターマーク宣言 |
