# バッチ設計書 17-BatchExecPythonGroupAggregate

## 概要

本ドキュメントは、Apache Flink Table Plannerにおける、Pythonユーザー定義集約関数（UDAF）を使用したグループ集約処理のバッチ実行ノードの設計を記述したものである。

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

BatchExecPythonGroupAggregateは、Pythonで実装されたユーザー定義集約関数（UDAF）を使用してグループ集約を実行するバッチ実行ノードである。Apache Arrowを使用してJavaとPython間でデータを効率的に転送する。

**業務上の目的・背景**：Flinkでは、Javaだけでなく、Pythonでもユーザー定義関数を実装できる。特にデータサイエンティストやPythonに精通した開発者にとって、Pythonで集約ロジックを記述できることは大きなメリットである。本ノードは、PandasやNumPyなどのPythonライブラリを活用した集約処理を可能にする。

**バッチの実行タイミング**：Flink SQLクエリの実行時、Python UDAFを使用するGROUP BY句を含むクエリにおいて呼び出される。PlannerがPython UDAFの使用を検出した場合に自動的に選択される。

**主要な処理内容**：
1. 入力ストリームからデータを受信
2. Apache Arrowフォーマットでデータをバッチ化
3. Pythonワーカープロセスにデータを転送
4. Python側でUDAFを実行
5. 結果をArrowフォーマットで受信
6. Java側で結果を処理し出力

**前後の処理との関連**：入力としてExchange（データパーティショニング）やCalc（フィルタ・射影）ノードを受け取る。Python処理のため、Pythonワーカープロセスが起動され、管理メモリ（ManagedMemory）を使用する場合がある。

**影響範囲**：Flink Table APIおよびSQL APIを使用するすべてのバッチジョブにおいて、Python UDAFを使用するクエリに影響する。

## バッチ種別

集計処理（Python UDAF）

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 随時（クエリ実行時） |
| 実行時刻 | N/A |
| 実行曜日 | N/A |
| 実行日 | N/A |
| トリガー | Flink SQLクエリの実行 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力データストリーム | 入力ストリームが存在すること |
| Python環境 | Pythonランタイムがインストールされていること |
| PyFlink | PyFlinkライブラリが利用可能であること |
| 集約関数 | Python UDAFが定義されていること |

### 実行可否判定

Flink PlannerがPython UDAFを含むクエリを検出した場合に自動的に選択される。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| grouping | int[] | Yes | N/A | グルーピングキーのフィールドインデックス |
| auxGrouping | int[] | Yes | N/A | 補助グルーピングキーのフィールドインデックス |
| aggCalls | AggregateCall[] | Yes | N/A | 集約関数呼び出しの配列（Python UDAF） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 入力ストリーム | RowData | 集約対象のデータストリーム |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 集約結果ストリーム | RowData | グルーピングキーと集約結果を含むレコード |

### 出力ファイル仕様

ファイル出力なし（ストリーム出力のみ）

## 処理フロー

### 処理シーケンス

```
1. 入力変換の取得
   └─ 入力ストリームをTransformationとして取得
2. Python設定の抽出
   └─ CommonPythonUtil.extractPythonConfiguration()でPython設定を取得
3. Python関数情報の抽出
   └─ extractPythonAggregateFunctionInfosFromAggregateCall()で関数情報を抽出
4. 投影コード生成
   └─ UdafInputProjection: UDAF入力フィールドの投影
   └─ GroupKey: グルーピングキーの投影
   └─ GroupSet: 補助グルーピングキーの投影
5. Pythonオペレーターの構築
   └─ BatchArrowPythonGroupAggregateFunctionOperatorを構築
6. 管理メモリの宣言
   └─ Pythonワーカーが管理メモリを使用する場合は宣言
7. Transformationの作成
   └─ オペレーターを使用してTransformationを作成
8. 結果の出力
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[入力ストリーム取得]
    B --> C[Python設定抽出]
    C --> D[Python関数情報抽出]
    D --> E[投影コード生成]
    E --> F[Pythonオペレーター構築]
    F --> G{管理メモリ使用?}
    G -->|Yes| H[ManagedMemoryUseCase.PYTHON宣言]
    G -->|No| I[Transformation作成]
    H --> I
    I --> J[Arrow形式でデータ転送]
    J --> K[Python UDAF実行]
    K --> L[結果受信]
    L --> M[結果出力]
    M --> N[バッチ終了]
```

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

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

本バッチノードはインメモリ処理であり、直接的なデータベース操作は行わない。

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| N/A | N/A | N/A | データベース操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | Pythonオペレーター構築失敗 | リフレクションエラー | Python環境とPyFlinkのバージョンを確認 |
| PythonException | Python実行時エラー | Python UDAF内でのエラー | UDAFのコードをデバッグ |
| OutOfMemoryError | メモリ不足 | Arrowバッファがメモリ超過 | Python管理メモリを増加 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | Flinkジョブ設定に依存 |
| リトライ間隔 | Flinkジョブ設定に依存 |
| リトライ対象エラー | 一時的なリソース不足 |

### 障害時対応

ジョブ失敗時は、Flink Checkpointから再開するか、ジョブを最初から再実行する。Pythonワーカーのエラーログも確認する。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | バッチ全体（Flinkジョブ単位） |
| コミットタイミング | ジョブ完了時 |
| ロールバック条件 | ジョブ失敗時 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | データサイズに依存 |
| 目標処理時間 | クエリ複雑度とデータサイズに依存（Pythonオーバーヘッドあり） |
| メモリ使用量上限 | Pythonワーカーの管理メモリ設定に依存 |

## 排他制御

同一ジョブ内で複数のタスクが並列実行される。各タスクは独立したPythonワーカープロセスを持ち、明示的な排他制御は不要。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | Transformation作成時 | ノード情報、Python関数リスト |
| 進捗ログ | 処理中 | 処理件数、Arrowバッチ数 |
| 終了ログ | 処理完了時 | 処理完了ステータス |
| エラーログ | エラー発生時 | エラー詳細、Pythonスタックトレース |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理時間 | ジョブ設定に依存 | Flink監視システム |
| Pythonワーカー状態 | プロセス異常 | Flink監視システム |

## 備考

- BatchArrowPythonGroupAggregateFunctionOperatorを使用（リフレクションで動的ロード）
- Apache Arrowフォーマットを使用してJava-Python間のデータ転送を効率化
- CommonPythonUtil.isPythonWorkerUsingManagedMemory()で管理メモリ使用の判定
- ProjectionCodeGeneratorで3種類の投影を生成：
  - UdafInputProjection: UDAF入力フィールド
  - GroupKey: グルーピングキー
  - GroupSet: 補助グルーピングキー
