# 機能設計書 77-PyFlink DataStream

## 概要

本ドキュメントは、Apache Flink の PyFlink DataStream API に関する機能設計書である。flink-python モジュールで提供される、Python から Flink の DataStream API を利用するためのラッパーライブラリについて記載する。

### 本機能の処理概要

PyFlink DataStream API は、Python から Flink の Java DataStream API を透過的に利用するためのラッパーレイヤーである。Py4J を通じて Java オブジェクトと通信し、Python ユーザーが Flink のストリーム処理機能を活用できるようにする。

**業務上の目的・背景**：
- Python ユーザーが Flink のストリーム処理を利用可能にする
- 機械学習パイプラインとの統合
- データサイエンティストへの Flink 機能提供

**機能の利用シーン**：
- Python UDF（ユーザー定義関数）を使用したストリーム処理
- 機械学習モデルのリアルタイム推論
- データ変換・フィルタリング処理

**主要な処理内容**：
1. StreamExecutionEnvironment による実行環境の構築
2. DataStream による変換処理（map, filter, key_by 等）
3. Python UDF のシリアライズと Java 側への転送
4. ジョブ実行とステート管理

**関連システム・外部連携**：
- Py4J（Python-Java ブリッジ）
- Apache Beam（Python UDF 実行）
- Java DataStream API

**権限による制御**：特になし

## 関連画面

本機能はバックエンドライブラリであり、直接関連する画面はない。

## 機能種別

クライアントライブラリ / ストリーム処理 API

## 入力仕様

### StreamExecutionEnvironment 設定パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| parallelism | int | No | 並列度 | >= 1 |
| max_parallelism | int | No | 最大並列度 | 1 <= x <= 32768 |
| buffer_timeout | int | No | バッファタイムアウト（ミリ秒） | >= -1 |
| checkpoint_interval | int | No | チェックポイント間隔（ミリ秒） | > 0 |
| python_executable | str | No | Python インタープリタパス | - |
| requirements_file_path | str | No | requirements.txt パス | - |

### DataStream 変換パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| func | Callable/Function | Yes | 変換関数 | callable または Function サブクラス |
| output_type | TypeInformation | No | 出力型情報 | - |
| key_selector | Callable/KeySelector | Yes (key_by) | キー抽出関数 | callable または KeySelector |
| key_type | TypeInformation | No | キー型情報 | - |

### 入力データソース

- from_collection(): Python リストからの入力
- from_source(): Source コネクタからの入力
- add_source(): SourceFunction からの入力
- create_input(): InputFormat からの入力

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| DataStream | DataStream | 変換後のストリーム |
| DataStreamSink | DataStreamSink | シンク操作の結果 |
| JobExecutionResult | JobExecutionResult | ジョブ実行結果 |
| JobClient | JobClient | 非同期実行のクライアント |

### 出力先

- print(): 標準出力
- sink_to(): Sink コネクタ
- add_sink(): SinkFunction
- execute_and_collect(): クライアントへのイテレータ

## 処理フロー

### 処理シーケンス

```
1. 環境構築
   └─ StreamExecutionEnvironment.get_execution_environment()
   └─ Py4J 経由で Java StreamExecutionEnvironment 取得

2. データソース設定
   └─ from_collection() / from_source() / add_source()
   └─ DataStream オブジェクト生成

3. 変換処理定義
   └─ map() / filter() / key_by() / reduce() 等
   └─ ProcessFunction を内部的に使用
   └─ Python UDF をシリアライズ

4. シンク設定
   └─ sink_to() / add_sink() / print()
   └─ DataStreamSink 生成

5. ジョブ実行
   └─ execute() / execute_async()
   └─ StreamGraph 生成
   └─ Java ランタイムで実行
```

### フローチャート

```mermaid
flowchart TD
    subgraph Python [Python Layer]
        A[StreamExecutionEnvironment]
        B[DataStream]
        C[KeyedStream]
        D[DataStreamSink]
    end

    subgraph Py4J [Py4J Bridge]
        E[Java Gateway]
    end

    subgraph Java [Java Layer]
        F[J_StreamExecutionEnvironment]
        G[J_DataStream]
        H[J_KeyedStream]
        I[PythonOperator]
    end

    A -->|wrap| E
    E -->|create| F
    B -->|wrap| E
    E -->|transform| G
    C -->|wrap| E
    E -->|keyBy| H
    D -->|wrap| E
    E -->|execute| I
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-77-01 | 最大並列度上限 | max_parallelism <= 32768 | set_max_parallelism()時 |
| BR-77-02 | バッファタイムアウト | -1: バッファフル時のみフラッシュ、0: 即時フラッシュ、正数: 定期フラッシュ | set_buffer_timeout()時 |
| BR-77-03 | Python バージョン | Python 3.7 以上必須 | 実行時 |
| BR-77-04 | Beam バージョン | Apache Beam 2.54.0 以上 2.61.0 以下 | UDF 実行時 |
| BR-77-05 | 出力型指定 | 出力型未指定時は pickle バイト配列 | map/filter 等の変換時 |

### 計算ロジック

特になし

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TypeError | 型エラー | func が callable でも Function サブクラスでもない | 適切な関数を指定 |
| Exception | プロジェクションエラー | Tuple 以外で project() 使用 | Tuple DataStream を使用 |
| ValueError | パーティション数エラー | num_partitions <= 0 | 正のパーティション数を設定 |

### リトライ仕様

ライブラリ側でのリトライ機能はない。チェックポイントによるフォールトトレランス機能を使用する。

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

本機能はトランザクション管理を行わない。exactly-once セマンティクスはチェックポイント機能で実現する。

## パフォーマンス要件

- Python UDF は Apache Beam ランナーで実行
- オペレーターチェーン最適化（PythonOperatorChainingOptimizer）
- Pickle シリアライゼーションによるデータ転送

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

- Python コードのシリアライズと転送
- クラスタ上での Python 実行環境の分離

## 備考

- Python UDF は Java 処理より遅い場合がある
- 大規模データの場合は Java API 推奨

---

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

### 推奨読解順序

#### Step 1: 実行環境を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | stream_execution_environment.py | `flink-python/pyflink/datastream/stream_execution_environment.py` | 実行環境ラッパー |

**主要処理フロー（stream_execution_environment.py）**:
- **49-59行目**: クラス定義と Javadoc - StreamExecutionEnvironment の役割説明
- **61-64行目**: コンストラクタ - Java オブジェクトのラップと serializer 設定
- **74-90行目**: set_parallelism() - 並列度設定
- **92-106行目**: set_max_parallelism() - 最大並列度設定（上限 32768）
- **167-183行目**: set_buffer_timeout() - バッファタイムアウト設定
- **234-268行目**: enable_checkpointing() - チェックポイント有効化
- **398-414行目**: add_python_file() - Python 依存ファイル追加
- **416-456行目**: set_python_requirements() - requirements.txt 設定
- **521-558行目**: set_python_executable() - Python インタープリタパス設定
- **609-622行目**: execute() - ジョブ実行（StreamGraph 生成）
- **624-637行目**: execute_async() - 非同期ジョブ実行
- **670-695行目**: get_execution_environment() - 静的ファクトリメソッド
- **777-792行目**: from_collection() - コレクションからの DataStream 生成

#### Step 2: DataStream 変換を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | data_stream.py | `flink-python/pyflink/datastream/data_stream.py` | データストリーム変換 |

**主要処理フロー（data_stream.py）**:
- **69-80行目**: DataStream クラス定義 - Java DataStream のラッパー
- **79-80行目**: コンストラクタ - _j_data_stream にJavaオブジェクト保持
- **274-314行目**: map() - MapProcessFunctionAdapter で ProcessFunction に変換
- **316-356行目**: flat_map() - FlatMapProcessFunctionAdapter で変換
- **358-414行目**: key_by() - AddKey ProcessFunction でキー付加、KeyedStream 生成
- **416-456行目**: filter() - FilterProcessFunctionAdapter で変換
- **474-494行目**: union() - 複数ストリームのマージ
- **528-535行目**: shuffle() - シャッフルパーティショニング
- **576-583行目**: rebalance() - ラウンドロビンパーティショニング
- **642-665行目**: process() - ProcessFunction の適用
- **812-820行目**: add_sink() - SinkFunction の追加
- **822-838行目**: sink_to() - Sink コネクタへの出力
- **840-868行目**: execute_and_collect() - 実行と結果収集

#### Step 3: KeyedStream を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | data_stream.py | `flink-python/pyflink/datastream/data_stream.py` | KeyedStream クラス |

**主要処理フロー（KeyedStream - data_stream.py）**:
- **1093-1112行目**: KeyedStream クラス定義 - DataStream を継承
- **1114-1154行目**: map() - MapKeyedProcessFunctionAdapter で KeyedProcessFunction に変換
- **1156-1196行目**: flat_map() - FlatMapKeyedProcessFunctionAdapter で変換
- **1198-1281行目**: reduce() - ReduceProcessKeyedProcessFunctionAdapter でリデュース
- **1283-1299行目**: filter() - FilterKeyedProcessFunctionAdapter で変換

#### Step 4: 関数インターフェースを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | functions.py | `flink-python/pyflink/datastream/functions.py` | UDF インターフェース |

**主要処理フロー（functions.py）**:
- **73-131行目**: KeyedStateStore - ステートアクセスインターフェース
- **134-199行目**: RuntimeContext - ランタイムコンテキスト（並列度、タスク名等）
- **202-210行目**: Function 基底クラス - open/close ライフサイクル
- **213-236行目**: MapFunction - 1対1変換
- **239-272行目**: CoMapFunction - 2ストリーム結合変換
- **275-300行目**: FlatMapFunction - 1対N変換

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

```
StreamExecutionEnvironment (Python)
    │
    ├─ get_execution_environment()
    │      └─ get_gateway().jvm.StreamExecutionEnvironment.getExecutionEnvironment()
    │
    ├─ from_collection(collection)
    │      └─ _from_collection()
    │             └─ PickleSerializer.serialize()
    │             └─ j_stream_execution_environment.createInput()
    │
    ├─ DataStream
    │      │
    │      ├─ map(func)
    │      │      └─ MapProcessFunctionAdapter(func)
    │      │             └─ process()
    │      │                    └─ _get_one_input_stream_operator()
    │      │                           └─ j_data_stream.transform()
    │      │
    │      ├─ key_by(key_selector)
    │      │      └─ AddKey ProcessFunction
    │      │             └─ process()
    │      │             └─ j_data_stream.keyBy()
    │      │                    └─ KeyedStream
    │      │
    │      └─ sink_to(sink)
    │             └─ j_data_stream.sinkTo()
    │                    └─ DataStreamSink
    │
    └─ execute(job_name)
           └─ _generate_stream_graph()
           └─ j_stream_execution_environment.execute()
                  └─ JobExecutionResult
```

### データフロー図

```
[Python Application]                [Py4J Gateway]              [Java Runtime]

        │                                │                           │
        │   StreamExecutionEnvironment   │                           │
        └───────────────────────────────▶│                           │
                                         │   getExecutionEnvironment │
                                         └──────────────────────────▶│
                                                                     │
                                         ◀───────────────────────────┤
        ◀────────────────────────────────┤                           │
        │                                │                           │
        │   from_collection(data)        │                           │
        └───────────────────────────────▶│                           │
                                         │  Pickle serialize         │
                                         │  createInput()            │
                                         └──────────────────────────▶│
                                                                     │
                                         ◀───────────────────────────┤
        DataStream                       │                           │
        ◀────────────────────────────────┤                           │
        │                                │                           │
        │   map(python_func)             │                           │
        └───────────────────────────────▶│                           │
                                         │  serialize UDF            │
                                         │  transform()              │
                                         └──────────────────────────▶│
                                                                     │
                                                           PythonOperator
                                                                     │
                                         ◀───────────────────────────┤
        DataStream                       │                           │
        ◀────────────────────────────────┤                           │
        │                                │                           │
        │   execute()                    │                           │
        └───────────────────────────────▶│                           │
                                         │   generateStreamGraph()   │
                                         │   execute(StreamGraph)    │
                                         └──────────────────────────▶│
                                                                     │
                                                              Job Execution
                                                        (Python UDF via Beam)
                                                                     │
                                         ◀───────────────────────────┤
        JobExecutionResult               │                           │
        ◀────────────────────────────────┤                           │
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| stream_execution_environment.py | `flink-python/pyflink/datastream/stream_execution_environment.py` | ソース | 実行環境ラッパー |
| data_stream.py | `flink-python/pyflink/datastream/data_stream.py` | ソース | DataStream/KeyedStream ラッパー |
| functions.py | `flink-python/pyflink/datastream/functions.py` | ソース | UDF インターフェース |
| state.py | `flink-python/pyflink/datastream/state.py` | ソース | ステートラッパー |
| window.py | `flink-python/pyflink/datastream/window.py` | ソース | ウィンドウ定義 |
| checkpoint_config.py | `flink-python/pyflink/datastream/checkpoint_config.py` | ソース | チェックポイント設定 |
| checkpointing_mode.py | `flink-python/pyflink/datastream/checkpointing_mode.py` | ソース | チェックポイントモード |
| execution_mode.py | `flink-python/pyflink/datastream/execution_mode.py` | ソース | 実行モード |
| output_tag.py | `flink-python/pyflink/datastream/output_tag.py` | ソース | サイドアウトプット |
| slot_sharing_group.py | `flink-python/pyflink/datastream/slot_sharing_group.py` | ソース | スロット共有グループ |
| state_backend.py | `flink-python/pyflink/datastream/state_backend.py` | ソース | ステートバックエンド |
| timerservice.py | `flink-python/pyflink/datastream/timerservice.py` | ソース | タイマーサービス |
| utils.py | `flink-python/pyflink/datastream/utils.py` | ソース | ユーティリティ |
| async_data_stream.py | `flink-python/pyflink/datastream/async_data_stream.py` | ソース | 非同期ストリーム |
