# バッチ設計書 39-BatchExecMLPredictTableFunction

## 概要

本ドキュメントは、Apache Flinkのバッチ処理における機械学習モデル予測を実行する`BatchExecMLPredictTableFunction`の設計仕様を定義するものです。

### 本バッチの処理概要

BatchExecMLPredictTableFunctionは、登録された機械学習モデルを使用してバッチデータに対する推論（予測）処理を実行するための実行ノードです。SQL構文のML_PREDICT関数に対応し、同期・非同期の両方の予測処理をサポートします。

**業務上の目的・背景**：機械学習モデルを本番環境でバッチデータに適用することは、レコメンデーション生成、不正検知、需要予測などの多くの業務シナリオで必要とされます。本バッチは、FlinkのTable API/SQLからシームレスに機械学習モデルによる推論を実行する機能を提供し、データパイプラインへのML統合を実現します。

**バッチの実行タイミング**：ML_PREDICT関数を含むSQLクエリの実行時に、クエリプランナーによって自動的に生成されます。

**主要な処理内容**：
1. モデル仕様（ModelSpec）からモデルプロバイダー（ModelProvider）の取得
2. 同期/非同期オプションに基づく予測関数（PredictFunction/AsyncPredictFunction）の取得
3. MLPredictCodeGeneratorによる予測コードの生成
4. 同期モード：MLPredictRunnerとProcessOperatorによる処理
5. 非同期モード：AsyncMLPredictRunnerとAsyncWaitOperatorによる処理
6. 入力特徴量（features）をモデルに渡して予測結果を取得
7. 入力行と予測結果を結合して出力

**前後の処理との関連**：前段として任意のBatchExecNodeからの入力データを受け取ります。後段としてBatchExecCalc、BatchExecSink等に接続されます。

**影響範囲**：ML_PREDICT関数を使用するすべてのバッチクエリに影響します。機械学習モデルの推論パフォーマンスがクエリ全体に影響します。

## バッチ種別

機械学習 / 推論処理

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | クエリ実行時（随時） |
| 実行時刻 | N/A |
| 実行曜日 | N/A |
| 実行日 | N/A |
| トリガー | ML_PREDICT関数使用時 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| モデル登録 | ML_PREDICT対象のモデルがカタログに登録されていること |
| PredictRuntimeProvider | モデルプロバイダーがPredictRuntimeProviderを実装していること（同期モード） |
| AsyncPredictRuntimeProvider | モデルプロバイダーがAsyncPredictRuntimeProviderを実装していること（非同期モード） |

### 実行可否判定

モデルプロバイダーが要求される予測関数（同期/非同期）を提供できる場合のみ実行可能。提供できない場合はTableExceptionがスローされる。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| mlPredictSpec | MLPredictSpec | Yes | - | 予測仕様（特徴量、ランタイム設定等） |
| modelSpec | ModelSpec | Yes | - | モデル仕様（モデル識別子、解決済みモデル等） |
| asyncOptions | FunctionCallUtil.AsyncOptions | No | null | 非同期オプション（null時は同期モード） |
| inputProperty | InputProperty | Yes | - | 入力プロパティ |
| outputType | RowType | Yes | - | 出力型（入力＋予測結果） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 上流ExecNode | RowData | 予測対象の入力レコード（特徴量を含む） |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 下流ExecNode | RowData | 入力レコード＋予測結果カラム |

### 出力ファイル仕様

N/A（メモリ内データ処理のため）

## 処理フロー

### 処理シーケンス

```
1. モデルプロバイダーの取得
   └─ modelSpec.getModelProvider(context)
2. 予測関数の取得
   └─ 同期: PredictRuntimeProvider.createPredictFunction
   └─ 非同期: AsyncPredictRuntimeProvider.createAsyncPredictFunction
3. 予測コード生成
   └─ 同期: MLPredictCodeGenerator.generateSyncPredictFunction
   └─ 非同期: MLPredictCodeGenerator.generateAsyncPredictFunction
4. コレクター生成
   └─ MLPredictCodeGenerator.generateCollector
5. オペレーター構築
   └─ 同期: ProcessOperator(MLPredictRunner)
   └─ 非同期: AsyncWaitOperatorFactory(AsyncMLPredictRunner)
6. OneInputTransformation構築
   └─ 並列度は入力と同一
7. 実行（ランタイム）
   └─ 入力行から特徴量抽出
   └─ モデルに予測要求
   └─ 結果を出力行に結合
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[ModelProvider取得]
    B --> C{非同期モード?}
    C -->|Yes| D[AsyncPredictFunction取得]
    C -->|No| E[PredictFunction取得]
    D --> F[generateAsyncPredictFunction]
    E --> G[generateSyncPredictFunction]
    F --> H[AsyncWaitOperatorFactory構築]
    G --> I[ProcessOperator構築]
    H --> J[OneInputTransformation構築]
    I --> J
    J --> K[入力行処理]
    K --> L[特徴量抽出]
    L --> M[モデル予測]
    M --> N[結果結合]
    N --> O[バッチ終了]
```

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

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

N/A（インメモリ処理のため）

### テーブル別操作詳細

N/A

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | プロバイダー不整合 | 要求された予測関数をプロバイダーが提供しない | モデル実装を確認 |
| RuntimeException | 予測エラー | モデル推論中にエラー発生 | モデルとデータを確認 |
| TimeoutException | タイムアウト | 非同期予測がタイムアウト | asyncTimeoutを調整 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | N/A（予測エラーは再試行設計に依存） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

### 障害時対応

モデルの状態、入力データの妥当性、非同期設定（タイムアウト、バッファ容量）を確認。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | N/A（ML推論） |
| コミットタイミング | N/A |
| ロールバック条件 | N/A |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 入力データ量に依存 |
| 目標処理時間 | モデル推論速度に依存 |
| メモリ使用量上限 | モデルサイズと非同期バッファに依存 |

## 排他制御

並列実行可能。各タスクは独立してモデル推論を実行。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| DEBUG | オペレーター生成時 | モデル識別子と設定情報 |
| ERROR | 予測エラー時 | エラーメッセージとスタックトレース |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 予測レイテンシ | アプリケーション定義 | Flinkメトリクス |
| 予測失敗率 | アプリケーション定義 | Flinkメトリクス |

## 備考

- Flink 2.3以降で利用可能（minPlanVersion = FlinkVersion.v2_3）
- @ExecNodeMetadataアノテーションあり（永続化対応）
- 消費する設定オプション:
  - `table.exec.async-ml-predict.max-concurrent-operations`
  - `table.exec.async-ml-predict.timeout`
  - `table.exec.async-ml-predict.output-mode`
- CommonExecMLPredictTableFunctionを継承
- SingleTransformationTranslatorを実装
- PipelineOptions.OBJECT_REUSEを考慮したコード生成
- ModelPredictRuntimeProviderContextでモデルコンテキストを渡す
- 非同期モードは高スループット向け、同期モードは低レイテンシ向け
