# 機能設計書 1-Function API

## 概要

本ドキュメントは、Apache Flinkのコア機能であるFunction APIについて詳細に記載する。Function APIはストリーム処理およびバッチ処理の基本関数インターフェースを提供し、ユーザーがデータ変換ロジックを定義するための基盤となる。

### 本機能の処理概要

Function APIは、Apache Flinkにおけるユーザー定義関数（UDF）の基盤インターフェースを提供する機能である。この機能により、開発者はmap、filter、reduce、aggregate等の変換処理を独自に実装し、ストリームデータやバッチデータに対して柔軟なデータ処理を行うことができる。

**業務上の目的・背景**：大規模データ処理において、データの変換・集約・フィルタリングは最も基本的かつ頻繁に行われる操作である。Function APIは、これらの操作を標準化されたインターフェースとして提供することで、開発者がビジネスロジックに集中できる環境を構築する。また、Java 8のラムダ式をサポートするSAM（Single Abstract Method）インターフェースとして設計されており、簡潔なコード記述を可能にしている。

**機能の利用シーン**：
- ETLパイプラインでのデータ変換処理
- リアルタイムストリーム処理でのデータフィルタリング
- 集計処理（合計、平均、カウント等）の実装
- 複数データストリームの結合処理

**主要な処理内容**：
1. MapFunction：1入力1出力の変換処理（データ型変換、フィールド抽出等）
2. FlatMapFunction：1入力から0個以上の出力を生成する変換処理
3. FilterFunction：条件に基づくデータのフィルタリング
4. ReduceFunction：同一キーの要素を結合して単一値に集約
5. AggregateFunction：中間状態（アキュムレータ）を持つ柔軟な集計処理

**関連システム・外部連携**：Function APIはFlink内部のオペレーター層と密接に連携し、TaskManagerによって分散実行される。また、RuntimeContextを通じてアキュムレーター、分散キャッシュ、ステート管理機能にアクセス可能である。

**権限による制御**：Function API自体には権限制御は存在しないが、RichFunction実装においてRuntimeContextを通じて実行時コンテキストにアクセスする際に、並列度やサブタスクインデックスに基づいた処理分岐が可能である。

## 関連画面

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

## 機能種別

計算処理 / データ変換

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| value | T（ジェネリック型） | Yes | 処理対象のデータ要素 | null許容はFunction実装に依存 |
| collector | Collector<O> | FlatMapのみ | 出力結果を収集するコレクター | Not null |
| accumulator | ACC | AggregateFunction | 中間集計状態 | 初期化必須 |

### 入力データソース

- DataStreamまたはDataSetからの要素
- 上流オペレーターからの変換済みデータ
- ソースコネクタからの読み込みデータ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| result | O（ジェネリック型） | 変換後のデータ要素 |

### 出力先

- 下流オペレーターへの入力
- DataStreamSinkまたはDataSetSink

## 処理フロー

### 処理シーケンス

```
1. 関数のインスタンス化
   └─ TaskManagerがUDFをデシリアライズして並列インスタンスを作成
2. ライフサイクル初期化（RichFunctionの場合）
   └─ open(OpenContext)メソッドでリソース初期化
3. 入力データ処理
   └─ 各入力要素に対してmap/flatMap/filter/reduce等のメソッドを呼び出し
4. ライフサイクル終了（RichFunctionの場合）
   └─ close()メソッドでリソース解放
```

### フローチャート

```mermaid
flowchart TD
    A[開始] --> B[Functionインスタンス化]
    B --> C{RichFunction?}
    C -->|Yes| D[open実行]
    C -->|No| E[データ処理開始]
    D --> E
    E --> F{次の要素あり?}
    F -->|Yes| G[変換処理実行]
    G --> H[結果出力]
    H --> F
    F -->|No| I{RichFunction?}
    I -->|Yes| J[close実行]
    I -->|No| K[終了]
    J --> K
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | シリアライズ必須 | すべてのFunctionはSerializableを実装する必要がある | 常時 |
| BR-002 | SAMインターフェース | 基本Functionはラムダ式で実装可能 | Java 8以降 |
| BR-003 | 不変性推奨 | FilterFunctionは入力要素を変更してはならない | FilterFunction使用時 |

### 計算ロジック

**ReduceFunction**: 2つの値を1つに結合
```
reduce(value1, value2) -> combined_value
```

**AggregateFunction**: アキュムレータパターン
```
createAccumulator() -> initial_accumulator
add(value, accumulator) -> updated_accumulator
getResult(accumulator) -> final_result
merge(acc1, acc2) -> merged_accumulator
```

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | InvalidTypesException | 型推論に失敗した場合 | TypeInformationを明示的に指定 |
| - | IllegalStateException | RuntimeContextが未初期化 | open()内またはその後でRuntimeContextにアクセス |
| - | Exception | UDF内で例外発生 | フェイルオーバー処理が実行される |

### リトライ仕様

Flinkのチェックポイント/リカバリ機構により、障害発生時は最後のチェックポイントから再実行される。

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

Function API自体にはトランザクション機能はない。チェックポイント機構と連携してExactly-once保証を実現する。

## パフォーマンス要件

- 各関数呼び出しはマイクロ秒オーダーで完了することが期待される
- ステートレス関数は並列度に応じてリニアにスケールする

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

- UDFはシリアライズされてTaskManagerに転送されるため、クラスパス汚染に注意
- 外部リソースへのアクセスはopen/closeライフサイクルで管理

## 備考

Function APIはFlink 1.xから存在する安定したAPIであり、@Publicアノテーションが付与されている。

---

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

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

### 推奨読解順序

#### Step 1: データ構造を理解する

まず、Function APIの基本インターフェース構造を理解することが重要である。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Function.java | `flink-core-api/src/main/java/org/apache/flink/api/common/functions/Function.java` | 全UDFの基底インターフェース。空のマーカーインターフェースでSerializableを継承 |
| 1-2 | RichFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java` | ライフサイクルメソッド（open/close）とRuntimeContextアクセスを定義 |

**読解のコツ**: Functionインターフェースは意図的に空（メソッドなし）として設計されている。これはJava 8のラムダ式サポートのためのSAM（Single Abstract Method）インターフェースを実現するためである。

#### Step 2: 基本的な変換関数を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | MapFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/MapFunction.java` | 1対1変換の基本パターン |
| 2-2 | FlatMapFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/FlatMapFunction.java` | Collectorを使用した1対多変換パターン |
| 2-3 | FilterFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/FilterFunction.java` | 述語関数パターン |

**主要処理フロー**:
1. **29-30行目（Function.java）**: @Publicアノテーションとjava.io.Serializableの継承を確認
2. **45-56行目（MapFunction.java）**: @FunctionalInterfaceとmap()メソッドの定義
3. **45-56行目（FlatMapFunction.java）**: Collector引数を持つflatMap()メソッド

#### Step 3: 集計関数を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ReduceFunction.java | `flink-core-api/src/main/java/org/apache/flink/api/common/functions/ReduceFunction.java` | 二項演算による集約パターン |
| 3-2 | AggregateFunction.java | `flink-core-api/src/main/java/org/apache/flink/api/common/functions/AggregateFunction.java` | アキュムレータを使用した分散集計パターン |

**主要処理フロー**:
- **51-64行目（ReduceFunction.java）**: reduce(T, T) -> T の二項演算インターフェース
- **114-160行目（AggregateFunction.java）**: 4つのメソッド（createAccumulator, add, getResult, merge）によるアキュムレータパターン

#### Step 4: RichFunction実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | AbstractRichFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/AbstractRichFunction.java` | RichFunctionの基本実装 |
| 4-2 | RuntimeContext.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java` | 実行時コンテキストへのアクセス |

**主要処理フロー**:
- **31-75行目（AbstractRichFunction.java）**: RuntimeContextの管理とopen/closeのデフォルト実装

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

```
DataStream/DataSet API
    │
    ├─ map(MapFunction)
    │      └─ StreamMap operator
    │             └─ MapFunction.map(T) -> O
    │
    ├─ flatMap(FlatMapFunction)
    │      └─ StreamFlatMap operator
    │             └─ FlatMapFunction.flatMap(T, Collector<O>)
    │
    ├─ filter(FilterFunction)
    │      └─ StreamFilter operator
    │             └─ FilterFunction.filter(T) -> boolean
    │
    └─ reduce(ReduceFunction)
           └─ ReduceOperator
                  └─ ReduceFunction.reduce(T, T) -> T
```

### データフロー図

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

DataSource ───▶ Function.method() ───▶ 下流オペレーター
     │                  │
     │                  ├─ MapFunction: 1対1変換
     │                  ├─ FlatMapFunction: 1対多変換
     │                  ├─ FilterFunction: フィルタリング
     │                  └─ ReduceFunction: 集約
     │
     └─ 入力型T         出力型O
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Function.java | `flink-core-api/src/main/java/org/apache/flink/api/common/functions/Function.java` | ソース | 基底インターフェース |
| RichFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java` | ソース | ライフサイクル管理インターフェース |
| AbstractRichFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/AbstractRichFunction.java` | ソース | RichFunctionの抽象実装 |
| MapFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/MapFunction.java` | ソース | Map変換インターフェース |
| FlatMapFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/FlatMapFunction.java` | ソース | FlatMap変換インターフェース |
| FilterFunction.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/FilterFunction.java` | ソース | Filter変換インターフェース |
| ReduceFunction.java | `flink-core-api/src/main/java/org/apache/flink/api/common/functions/ReduceFunction.java` | ソース | Reduce集約インターフェース |
| AggregateFunction.java | `flink-core-api/src/main/java/org/apache/flink/api/common/functions/AggregateFunction.java` | ソース | Aggregate集約インターフェース |
| RuntimeContext.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java` | ソース | 実行時コンテキスト |
| OpenContext.java | `flink-core/src/main/java/org/apache/flink/api/common/functions/OpenContext.java` | ソース | 初期化コンテキスト |
