# 機能設計書 83-Table例

## 概要

本ドキュメントは、Apache FlinkのTable API/SQLサンプルプログラム（flink-examples-table）の機能設計書である。開発者がFlink Table APIおよびSQL機能の使用方法を学ぶためのリファレンス実装を提供する。

### 本機能の処理概要

flink-examples-tableモジュールは、Apache Flink Table API/SQLの使用方法を示すサンプルプログラム集である。リレーショナルクエリ処理、テーブル変換、ユーザー定義関数（UDF）、カスタムコネクタの実装例を提供する。

**業務上の目的・背景**：Table APIとSQLは、Flinkの宣言的APIであり、リレーショナルデータ処理をシンプルに記述できる。SQLに慣れた開発者がFlinkを利用する際のエントリーポイントとなる。サンプルコードを通じて、Table APIの使い方、SQLクエリの記述方法、カスタム関数・コネクタの実装方法を学習できる。

**機能の利用シーン**：(1)Table API学習時のリファレンス、(2)SQLクエリ記述方法の学習、(3)UDF実装のテンプレート、(4)カスタムコネクタ開発のガイド、(5)機能検証・デモンストレーション

**主要な処理内容**：
1. **GettingStartedExample**: Table API基本操作のチュートリアル
2. **WordCountSQLExample**: SQLによる単語カウント
3. **StreamSQLExample**: ストリームデータへのSQL適用
4. **StreamWindowSQLExample**: ウィンドウSQLの使用例
5. **TemporalJoinSQLExample**: テンポラル結合の実装例
6. **UpdatingTopCityExample**: 更新ストリームの処理例
7. **AdvancedFunctionsExample**: 高度なUDFの実装例
8. **ChangelogSocketExample**: カスタムコネクタの実装例

**関連システム・外部連携**：ソケットストリーム（カスタムコネクタ経由）

**権限による制御**：なし（サンプルプログラム）

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | サンプルプログラムのため関連画面なし |

## 機能種別

サンプルコード / 学習リソース / リファレンス実装

## 入力仕様

### 入力パラメータ

GettingStartedExampleは引数不要で実行可能。サンプルデータはコード内に定義されている。

### 入力データソース

- インラインデータ（env.fromValues()）
- ソケットストリーム（ChangelogSocketExample）

## 出力仕様

### 出力データ

サンプルにより異なる。GettingStartedExampleでは顧客情報のフィルタリング・変換結果を出力。

### 出力先

- 標準出力（execute().print()）
- コレクション（execute().collect()）

## 処理フロー

### 処理シーケンス（GettingStartedExample）

```
1. TableEnvironment作成
   └─ TableEnvironment.create(EnvironmentSettings.inBatchMode())
2. サンプルデータからテーブル作成
   └─ env.fromValues(Row.of(...))
3. テーブル変換
   ├─ select() - 列選択
   └─ as() - 列名付与
4. ビュー登録
   └─ env.createTemporaryView("customers", table)
5. SQLクエリ実行
   └─ env.sqlQuery("SELECT COUNT(*) ...")
6. Table APIによる追加変換
   └─ filter(), select(), call()
7. 結果収集
   └─ execute().collect()
```

### フローチャート

```mermaid
flowchart TD
    A[main開始] --> B[EnvironmentSettings設定]
    B --> C[TableEnvironment作成]
    C --> D[サンプルデータ定義]
    D --> E[fromValues - テーブル作成]
    E --> F[select/as - 列変換・命名]
    F --> G[createTemporaryView - ビュー登録]
    G --> H[sqlQuery - SQLクエリ実行]
    H --> I[execute().print() - 結果表示]
    I --> J[filter/select/call - Table API変換]
    J --> K[execute().collect() - 結果収集]
    K --> L[結果検証]
    L --> M[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | バッチモード実行 | GettingStartedExampleはバッチモードで実行 | デフォルト設定 |
| BR-002 | UDF登録 | ScalarFunctionを継承したクラスはcall()で呼び出し可能 | UDF使用時 |

### 計算ロジック

- AddressNormalizer: 住所文字列を大文字変換、非単語文字をスペースに置換、連続スペースを1つに正規化

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

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

本機能はデータベースを使用しない。

## エラー処理

### エラーケース一覧

サンプルプログラムのため特別なエラー処理は含まれていない。

### リトライ仕様

なし

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

なし

## パフォーマンス要件

サンプルプログラムのため特定のパフォーマンス要件はない。

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

サンプルプログラムのため特別なセキュリティ対策は含まれていない。

## 備考

- Table APIとSQLの両方のスタイルを1つのプログラムで組み合わせ可能
- カスタムコネクタ実装例（SocketDynamicTableFactory）も含まれる

---

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

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

### 推奨読解順序

#### Step 1: Table API基本操作を理解する

GettingStartedExampleで、Table APIの基本的な使用パターンを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | GettingStartedExample.java | `flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/basics/GettingStartedExample.java` | Table API基本パターン |

**主要処理フロー**:
- **63-65行目**: EnvironmentSettings.inBatchMode()でバッチモード設定、TableEnvironment作成
- **68-135行目**: env.fromValues()でサンプルデータからテーブル作成
- **138行目**: withColumns(range(1, 7))で列選択
- **141-149行目**: as()で列名付与
- **152行目**: createTemporaryView()でビュー登録
- **156-162行目**: sqlQuery()でSQLクエリ実行、execute().print()で結果表示
- **166-175行目**: filter(), select(), call()でTable API変換
- **179行目**: execute().collect()で結果収集

**読解のコツ**: Table APIはメソッドチェーンでSQLライクな操作を記述する。fromValues()→変換→execute()が基本パターン。

#### Step 2: ユーザー定義関数を理解する

ScalarFunctionを継承したUDFの実装方法を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | GettingStartedExample.java | `flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/basics/GettingStartedExample.java` | AddressNormalizerクラス（213-224行目） |

**主要処理フロー**:
- **213行目**: ScalarFunction継承
- **217行目**: eval()メソッドで入出力定義（リフレクションで型推論）
- **221-223行目**: 文字列正規化ロジック

#### Step 3: カスタムコネクタを理解する

DynamicTableSourceの実装方法を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SocketDynamicTableFactory.java | `flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/connectors/SocketDynamicTableFactory.java` | コネクタファクトリー |
| 3-2 | SocketDynamicTableSource.java | `flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/connectors/SocketDynamicTableSource.java` | テーブルソース |

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

```
GettingStartedExample.main()
    │
    ├─ EnvironmentSettings.newInstance().inBatchMode().build()
    │
    ├─ TableEnvironment.create(settings)
    │
    ├─ env.fromValues(Row.of(...))
    │      └─ Table作成
    │
    ├─ table.select(withColumns(range(1, 7)))
    │      └─ 列選択
    │
    ├─ table.as("name", "date_of_birth", ...)
    │      └─ 列名付与
    │
    ├─ env.createTemporaryView("customers", table)
    │      └─ ビュー登録
    │
    ├─ env.sqlQuery("SELECT COUNT(*) ...")
    │      └─ SQLクエリ実行
    │
    ├─ env.from("customers")
    │      .filter($("gender").isNotNull())
    │      .filter($("has_newsletter").isEqual(true))
    │      .select(...)
    │      .call(AddressNormalizer.class, ...)
    │
    └─ table.execute().collect()
           └─ 結果収集
```

### データフロー図

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

インラインデータ        fromValues()
(Row.of(...))    ────▶ ┌──────────────────┐
                       │ Table作成         │
                       └──────────────────┘
                              │
                       select/as
                       ┌──────────────────┐
                       │ 列選択・命名      │
                       └──────────────────┘
                              │
                       createTemporaryView
                       ┌──────────────────┐
                       │ ビュー登録        │
                       └──────────────────┘
                              │
                       sqlQuery / Table API
                       ┌──────────────────┐
                       │ クエリ実行        │──▶ execute().print()
                       │ filter/select    │    execute().collect()
                       │ call(UDF)        │
                       └──────────────────┘
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| GettingStartedExample.java | `flink-examples/flink-examples-table/src/main/java/.../basics/GettingStartedExample.java` | ソース | Table API入門サンプル |
| WordCountSQLExample.java | `flink-examples/flink-examples-table/src/main/java/.../basics/WordCountSQLExample.java` | ソース | SQL版WordCount |
| StreamSQLExample.java | `flink-examples/flink-examples-table/src/main/java/.../basics/StreamSQLExample.java` | ソース | ストリームSQL |
| StreamWindowSQLExample.java | `flink-examples/flink-examples-table/src/main/java/.../basics/StreamWindowSQLExample.java` | ソース | ウィンドウSQL |
| TemporalJoinSQLExample.java | `flink-examples/flink-examples-table/src/main/java/.../basics/TemporalJoinSQLExample.java` | ソース | テンポラル結合 |
| UpdatingTopCityExample.java | `flink-examples/flink-examples-table/src/main/java/.../basics/UpdatingTopCityExample.java` | ソース | 更新ストリーム処理 |
| AdvancedFunctionsExample.java | `flink-examples/flink-examples-table/src/main/java/.../functions/AdvancedFunctionsExample.java` | ソース | 高度なUDF |
| SocketDynamicTableFactory.java | `flink-examples/flink-examples-table/src/main/java/.../connectors/SocketDynamicTableFactory.java` | ソース | カスタムコネクタファクトリー |
| SocketDynamicTableSource.java | `flink-examples/flink-examples-table/src/main/java/.../connectors/SocketDynamicTableSource.java` | ソース | カスタムテーブルソース |
| ChangelogSocketExample.java | `flink-examples/flink-examples-table/src/main/java/.../connectors/ChangelogSocketExample.java` | ソース | カスタムコネクタ使用例 |
