# バッチ設計書 18-BatchExecPythonGroupWindowAggregate

## 概要

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

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

BatchExecPythonGroupWindowAggregateは、Pythonで実装されたユーザー定義集約関数（UDAF）を使用してウィンドウ集約を実行するバッチ実行ノードである。タンブリング、ホッピングなどのウィンドウ集約にPython UDAFを適用する。

**業務上の目的・背景**：ウィンドウ集約とPython UDAFの組み合わせにより、時系列データに対してPythonの豊富なライブラリ（Pandas、NumPy、SciPyなど）を活用した高度な分析が可能になる。例えば、5分ウィンドウごとの統計分析や異常検知などに使用できる。

**バッチの実行タイミング**：Flink SQLクエリの実行時、Python UDAFを使用するGROUP BY TUMBLE()/HOP()などのウィンドウ関数を含むクエリにおいて呼び出される。

**主要な処理内容**：
1. 入力ストリームからデータを受信
2. 時間フィールドに基づいてウィンドウを割り当て
3. Apache Arrowフォーマットでデータをバッチ化
4. Pythonワーカープロセスにデータを転送
5. Python側でウィンドウごとにUDAFを実行
6. 結果をArrowフォーマットで受信
7. ウィンドウプロパティ（開始、終了、行時刻）を追加して出力

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

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

## バッチ種別

集計処理（Pythonウィンドウ集約）

## 実行スケジュール

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

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力データストリーム | 入力ストリームが存在すること |
| Python環境 | Pythonランタイムがインストールされていること |
| ウィンドウ定義 | LogicalWindowでウィンドウタイプとサイズが定義されていること |
| 時間フィールド | 入力に時間フィールドが存在すること |

### 実行可否判定

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

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| grouping | int[] | Yes | N/A | グルーピングキーのフィールドインデックス |
| auxGrouping | int[] | Yes | N/A | 補助グルーピングキーのフィールドインデックス |
| aggCalls | AggregateCall[] | Yes | N/A | 集約関数呼び出しの配列（Python UDAF） |
| window | LogicalWindow | Yes | N/A | ウィンドウ定義 |
| inputTimeFieldIndex | int | Yes | N/A | 入力時間フィールドのインデックス |
| namedWindowProperties | NamedWindowProperty[] | Yes | N/A | ウィンドウプロパティ |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 入力ストリーム | RowData | 集約対象のデータストリーム（時間フィールドを含む） |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 集約結果ストリーム | RowData | グルーピングキー、ウィンドウプロパティ、集約結果を含むレコード |

### 出力ファイル仕様

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

## 処理フロー

### 処理シーケンス

```
1. 入力変換の取得
   └─ 入力ストリームをTransformationとして取得
2. ウィンドウ定義の取得
   └─ WindowCodeGenerator.getWindowDef()でウィンドウサイズとスライドサイズを取得
3. Python設定の抽出
   └─ CommonPythonUtil.extractPythonConfiguration()でPython設定を取得
4. ウィンドウプロパティタイプの変換
   └─ WindowStart=0, WindowEnd=1, RowtimeAttribute=2に変換
5. Python関数情報の抽出
   └─ extractPythonAggregateFunctionInfosFromAggregateCall()で関数情報を抽出
6. 投影コード生成
   └─ UdafInputProjection, GroupKey, GroupSetの投影を生成
7. Pythonオペレーターの構築
   └─ BatchArrowPythonGroupWindowAggregateFunctionOperatorを構築
8. 管理メモリの宣言
   └─ Pythonワーカーが管理メモリを使用する場合は宣言
9. Transformationの作成
10. 結果の出力
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[入力ストリーム取得]
    B --> C[ウィンドウサイズ/スライドサイズ取得]
    C --> D[Python設定抽出]
    D --> E[ウィンドウプロパティタイプ変換]
    E --> F[Python関数情報抽出]
    F --> G[投影コード生成]
    G --> H[Pythonオペレーター構築]
    H --> I{管理メモリ使用?}
    I -->|Yes| J[ManagedMemoryUseCase.PYTHON宣言]
    I -->|No| K[Transformation作成]
    J --> K
    K --> L[ウィンドウ割り当て]
    L --> M[Arrow形式でデータ転送]
    M --> N[Python UDAF実行]
    N --> O[結果受信]
    O --> P[ウィンドウプロパティ追加]
    P --> Q[結果出力]
    Q --> R[バッチ終了]
```

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

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

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | Pythonオペレーター構築失敗 | リフレクションエラー | Python環境とPyFlinkのバージョンを確認 |
| TableException | 未サポートウィンドウプロパティ | WindowStart/End/Rowtime以外 | サポートされるプロパティを使用 |
| PythonException | Python実行時エラー | Python UDAF内でのエラー | UDAFのコードをデバッグ |

### リトライ仕様

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

### 障害時対応

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

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

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

## パフォーマンス要件

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

## 排他制御

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

## ログ出力

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

## 監視・アラート

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

## 備考

- BatchArrowPythonGroupWindowAggregateFunctionOperatorを使用（リフレクションで動的ロード）
- ウィンドウプロパティタイプの変換：
  - WindowStart → 0
  - WindowEnd → 1
  - RowtimeAttribute → 2
- TABLE_EXEC_WINDOW_AGG_BUFFER_SIZE_LIMITでバッファサイズを制御
- 3種類の投影を生成：UdafInputProjection、GroupKey、GroupSet
- CommonPythonUtil.isPythonWorkerUsingManagedMemory()で管理メモリ使用を判定
