# 機能設計書 7-DataStream変換

## 概要

本ドキュメントは、Apache FlinkのDataStream変換機能について詳細に記載する。DataStream変換はmap、flatMap、filter、keyBy、window等の変換操作を提供し、ストリームデータの変換・集約・フィルタリングを実現する中核機能である。

### 本機能の処理概要

DataStreamは同一型の要素からなるストリームを表現するクラスであり、様々な変換操作を適用することで別のDataStreamに変換できる。このクラスはストリーム処理の基本的な変換パターン（map、filter等）から高度な操作（keyBy、window、join等）まで幅広いAPIを提供する。

**業務上の目的・背景**：ストリーム処理において、データの変換・フィルタリング・集約は最も基本的かつ頻繁に行われる操作である。DataStreamクラスはこれらの操作を宣言的なAPIとして提供し、開発者が直感的にストリーム処理パイプラインを構築できるようにする。また、内部的にはTransformationオブジェクトとして変換を表現し、最終的にStreamGraphへと変換される。各変換操作はFlinkの型システムと連携し、型安全な処理を保証する。

**機能の利用シーン**：
- ETLパイプラインでのデータ変換
- リアルタイムフィルタリング
- キーベースの集計処理
- ストリーム結合（Union、Join）

**主要な処理内容**：
1. map()：1対1の要素変換
2. flatMap()：1対多の要素変換
3. filter()：条件に基づくフィルタリング
4. keyBy()：キーによるストリーム分割
5. union()：複数ストリームの結合
6. connect()：2つのストリームの接続
7. window()：ウィンドウ処理の適用

**関連システム・外部連携**：各変換操作はStreamOperatorとして実装され、TaskManagerで分散実行される。Transformationオブジェクトを通じてStreamGraphGeneratorと連携する。

**権限による制御**：DataStream変換自体には権限制御は存在しない。

## 関連画面

本機能はバックエンドのデータ処理機能であり、直接関連する画面は存在しない。ジョブ概要画面でDAGグラフとして変換処理の流れを確認可能。

## 機能種別

データ変換

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| function | MapFunction/FilterFunction等 | Yes | 変換関数 | Not null |
| keySelector | KeySelector | keyByのみ | キー抽出関数 | Not null |
| windowAssigner | WindowAssigner | windowのみ | ウィンドウ割り当て | Not null |

### 入力データソース

- 上流DataStreamからの要素
- DataStreamSourceからの初期データ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| DataStream<O> | DataStream | 変換後のデータストリーム |
| KeyedStream<T, K> | KeyedStream | キー付きデータストリーム |
| WindowedStream | WindowedStream | ウィンドウ化されたストリーム |

### 出力先

- 下流の変換操作
- DataStreamSink

## 処理フロー

### 処理シーケンス

```
1. 変換メソッド呼び出し
   └─ map()/filter()等のAPI呼び出し
2. オペレーター生成
   └─ StreamMap/StreamFilter等のオペレーター生成
3. Transformation生成
   └─ OneInputTransformation等として変換を登録
4. 新DataStream生成
   └─ 変換結果を表す新しいDataStreamを返却
```

### フローチャート

```mermaid
flowchart TD
    A[DataStream<T>] --> B{変換操作}
    B -->|map| C[StreamMap]
    B -->|flatMap| D[StreamFlatMap]
    B -->|filter| E[StreamFilter]
    B -->|keyBy| F[KeyedStream]
    C --> G[DataStream<O>]
    D --> G
    E --> G
    F --> H{ウィンドウ?}
    H -->|Yes| I[WindowedStream]
    H -->|No| J[KeyedStream操作]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | 型推論 | 変換関数から出力型を自動推論 | 常時 |
| BR-002 | クロージャクリーニング | ラムダ式のシリアライズ可能化 | ExecutionConfig有効時 |
| BR-003 | パーティショニング | keyBy後は同一キーが同一パーティション | keyBy後 |

### 計算ロジック

特になし。各変換関数のロジックはユーザー定義。

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | InvalidTypesException | 型推論失敗 | returns()で型を明示 |
| - | Exception | 変換関数内で例外 | フェイルオーバー処理 |

### リトライ仕様

チェックポイントが有効な場合、障害時は最後のチェックポイントから再実行。

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

DataStream変換自体にはトランザクション機能はない。

## パフォーマンス要件

- 変換操作はストリーミング処理のためレイテンシ最小化
- keyByはハッシュベースのパーティショニングで効率的に分散

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

- 変換関数はシリアライズされてTaskManagerに転送される

## 備考

DataStreamは@Publicアノテーションが付与された安定したAPIである。

---

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

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

### 推奨読解順序

#### Step 1: 基本クラスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | DataStream.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java` | DataStreamのメインクラス |

**読解のコツ**:
- **99-111行目**: JavaDocでDataStreamの役割を理解
- **111行目**: @Publicアノテーションとクラス定義
- **113行目**: StreamExecutionEnvironment保持
- **115行目**: Transformation保持
- map()、flatMap()、filter()メソッドの実装
- keyBy()メソッドの実装

#### Step 2: キー付きストリームを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KeyedStream.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java` | キー付きストリーム |

#### Step 3: オペレーターを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StreamMap.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java` | Mapオペレーター |
| 3-2 | StreamFilter.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java` | Filterオペレーター |

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

```
DataStream<T>
    │
    ├─ map(MapFunction)
    │      └─ StreamMap → OneInputTransformation → DataStream<O>
    │
    ├─ flatMap(FlatMapFunction)
    │      └─ StreamFlatMap → OneInputTransformation → DataStream<O>
    │
    ├─ filter(FilterFunction)
    │      └─ StreamFilter → OneInputTransformation → DataStream<T>
    │
    └─ keyBy(KeySelector)
           └─ PartitionTransformation → KeyedStream<T, K>
                  │
                  └─ window(WindowAssigner)
                         └─ WindowedStream
```

### データフロー図

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

DataStream<T> ───▶ map/filter/flatMap ───▶ DataStream<O>
     │                    │
     │                    └─ StreamOperator
     │
     └─ keyBy ───▶ KeyedStream<T, K>
                        │
                        └─ window ───▶ WindowedStream
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| DataStream.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java` | ソース | データストリーム |
| KeyedStream.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java` | ソース | キー付きストリーム |
| SingleOutputStreamOperator.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java` | ソース | 単一出力オペレーター |
| StreamMap.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java` | ソース | Mapオペレーター |
| StreamFilter.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java` | ソース | Filterオペレーター |
| StreamFlatMap.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java` | ソース | FlatMapオペレーター |
| OneInputTransformation.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java` | ソース | 変換定義 |
| PartitionTransformation.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java` | ソース | パーティション変換 |
