# 機能設計書 80-PyFlinkフォーマット

## 概要

本ドキュメントは、Apache Flink の PyFlink フォーマットに関する機能設計書である。flink-python モジュールで提供される、Python から Flink の各種データフォーマット（JSON, CSV, Avro, Parquet 等）を利用するためのラッパーライブラリについて記載する。

### 本機能の処理概要

PyFlink フォーマットは、データのシリアライズ/デシリアライズを担当するフォーマットライブラリである。SerializationSchema/DeserializationSchema インターフェースを通じて、バイナリデータと構造化データの相互変換を提供する。

**業務上の目的・背景**：
- 各種データフォーマットへの変換
- コネクタとの組み合わせによるデータ入出力
- 型安全なシリアライズ/デシリアライズ

**機能の利用シーン**：
- Kafka からの JSON メッセージ読み取り
- CSV ファイルの読み書き
- Avro/Parquet 形式でのファイル保存
- データ型変換

**主要な処理内容**：
1. SerializationSchema による構造化データからバイナリへの変換
2. DeserializationSchema によるバイナリから構造化データへの変換
3. BulkWriterFactory によるバルク書き込み
4. StreamFormat によるストリーム読み取り

**関連システム・外部連携**：
- PyFlink コネクタ（Kafka, FileSystem 等）
- Java フォーマットライブラリ

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

## 関連画面

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

## 機能種別

フォーマットライブラリ / シリアライズ・デシリアライズ

## 入力仕様

### JSON フォーマットパラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| type_info | TypeInformation | Yes* | 型情報 | type_info または json_schema |
| json_schema | str | Yes* | JSON スキーマ | type_info または json_schema |
| fail_on_missing_field | bool | No | 欠落フィールドでエラー | デフォルト false |
| ignore_parse_errors | bool | No | パースエラー無視 | デフォルト false |

### CSV フォーマットパラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| schema | CsvSchema | Yes | CSV スキーマ | - |
| separator | str | No | 配列要素セパレータ | デフォルト ; |

### Avro フォーマットパラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| schema | AvroSchema | Yes* | Avro スキーマ | schema または record_class |
| record_class | str | Yes* | レコードクラス名 | schema または record_class |

### 入力データソース

- バイナリデータ（byte[]）
- 構造化データ（Row, GenericRecord 等）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| SerializationSchema | SerializationSchema | シリアライズスキーマ |
| DeserializationSchema | DeserializationSchema | デシリアライズスキーマ |
| BulkWriterFactory | BulkWriterFactory | バルクライター |
| StreamFormat | StreamFormat | ストリーム読み取りフォーマット |

### 出力先

- コネクタへのシリアライズ済みデータ
- 構造化データ（Row, GenericRecord 等）

## 処理フロー

### 処理シーケンス

```
1. スキーマ定義
   └─ Builder パターンでスキーマ構築
   └─ 型情報またはスキーマ文字列から生成

2. フォーマット生成
   └─ build() で Java オブジェクト生成
   └─ Python ラッパーでラップ

3. コネクタへの接続
   └─ KafkaSource.set_value_only_deserializer()
   └─ FileSink.for_bulk_format()
```

### フローチャート

```mermaid
flowchart TD
    subgraph Schemas [Schema Types]
        A[SimpleStringSchema]
        B[JsonRowSchema]
        C[CsvSchema]
        D[AvroSchema]
    end

    subgraph Interfaces [Interfaces]
        E[SerializationSchema]
        F[DeserializationSchema]
        G[BulkWriterFactory]
        H[StreamFormat]
    end

    subgraph Connectors [Connectors]
        I[KafkaSource/Sink]
        J[FileSource/Sink]
    end

    A --> E
    A --> F
    B --> E
    B --> F
    C --> E
    C --> F
    D --> G
    D --> H
    E --> I
    F --> I
    G --> J
    H --> J
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-80-01 | 文字セット | SimpleStringSchema はデフォルト UTF-8 | 文字列シリアライズ時 |
| BR-80-02 | JSON 欠落フィールド | fail_on_missing_field=false で null 設定 | JSON デシリアライズ時 |
| BR-80-03 | CSV 配列セパレータ | デフォルト ; | CSV フォーマット時 |
| BR-80-04 | Parquet バッチサイズ | デフォルト 2048 | Parquet 読み取り時 |
| BR-80-05 | RowFieldExtractor | 指定インデックスは非負 | Row フィールド抽出時 |

### 計算ロジック

特になし

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TypeError | 型エラー | 必須パラメータ未設定 | パラメータを設定 |
| ValueError | 値エラー | field_index が負数 | 非負の値を指定 |
| IOError | パースエラー | JSON/CSV パース失敗 | データ形式を確認 |
| SchemaParseException | スキーマエラー | Avro スキーマ解析失敗 | スキーマを確認 |

### リトライ仕様

フォーマットライブラリ自体にはリトライ機能はない。

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

本機能はトランザクション管理を行わない。

## パフォーマンス要件

- BulkWriterFactory によるバッチ書き込み最適化
- Parquet のカラムナフォーマットによる効率的読み取り
- Avro のスキーマベースシリアライズ

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

- デシリアライズ時の入力検証

## 備考

- フォーマットは 1.16.0 以降で大幅に拡充
- コネクタと組み合わせて使用

---

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

### 推奨読解順序

#### Step 1: 基本シリアライズスキーマを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | serialization.py | `flink-python/pyflink/common/serialization.py` | シリアライズ基底クラス |

**主要処理フロー（serialization.py）**:
- **33-41行目**: SerializationSchema 基底クラス - シリアライズスキーマ
- **44-55行目**: DeserializationSchema 基底クラス - デシリアライズスキーマ
- **58-72行目**: SimpleStringSchema - 文字列シリアライズ（デフォルト UTF-8）
- **75-87行目**: ByteArraySchema - バイト配列シリアライズ
- **90-107行目**: Encoder - ファイルシンク用エンコーダ
- **110-117行目**: BulkWriterFactory - バルクライターファクトリ基底
- **120-131行目**: RowDataBulkWriterFactory - RowData 用バルクライター
- **134-181行目**: RowFieldExtractorSchema - Row フィールド抽出スキーマ

#### Step 2: JSON フォーマットを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | json.py | `flink-python/pyflink/datastream/formats/json.py` | JSON フォーマット |

**主要処理フロー（json.py）**:
- **28-107行目**: JsonRowDeserializationSchema - JSON デシリアライズ
- **39-44行目**: builder() - ビルダー取得
- **46-107行目**: Builder クラス
- **56-64行目**: type_info() - 型情報設定
- **66-77行目**: json_schema() - JSON スキーマ設定
- **79-85行目**: fail_on_missing_field() - 欠落フィールド時エラー
- **87-93行目**: ignore_parse_errors() - パースエラー無視
- **95-107行目**: build() - スキーマビルド
- **110-150行目**: JsonRowSerializationSchema - JSON シリアライズ
- **132-140行目**: with_type_info() - 型情報設定

#### Step 3: CSV フォーマットを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | csv.py | `flink-python/pyflink/datastream/formats/csv.py` | CSV フォーマット |

**主要処理フロー（csv.py）**:
- **39-69行目**: CsvSchema - CSV スキーマ
- **52-57行目**: builder() - ビルダー取得
- **72-200行目**: CsvSchemaBuilder - スキーマビルダー
- **86-91行目**: build() - スキーマビルド
- **93-111行目**: add_array_column() - 配列カラム追加
- **113-122行目**: add_boolean_column() - ブールカラム追加
- **124-138行目**: add_number_column() - 数値カラム追加
- **140-149行目**: add_string_column() - 文字列カラム追加
- **170-175行目**: set_allow_comments() - コメント許可設定
- **195-199行目**: set_array_element_separator() - 配列セパレータ設定

#### Step 4: Avro フォーマットを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | avro.py | `flink-python/pyflink/datastream/formats/avro.py` | Avro フォーマット |

**主要処理フロー（avro.py）**:
- **39-77行目**: AvroSchema - Avro スキーマ
- **55-64行目**: parse_string() - JSON 文字列からパース
- **66-77行目**: parse_file() - ファイルからパース
- **80-98行目**: GenericRecordAvroTypeInfo - GenericRecord 型情報
- **101-129行目**: AvroInputFormat - Avro 入力フォーマット
- **132-175行目**: AvroBulkWriters - Avro バルクライタービルダー
- **140-175行目**: for_generic_record() - GenericRecord 用ライター
- **178-200行目**: AvroRowDeserializationSchema - Avro Row デシリアライズ

#### Step 5: Parquet フォーマットを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | parquet.py | `flink-python/pyflink/datastream/formats/parquet.py` | Parquet フォーマット |

**主要処理フロー（parquet.py）**:
- **38-73行目**: AvroParquetReaders - Avro Parquet リーダー
- **46-73行目**: for_generic_record() - GenericRecord 用リーダー
- **76-117行目**: AvroParquetWriters - Avro Parquet ライター
- **84-117行目**: for_generic_record() - GenericRecord 用ライター
- **120-150行目**: ParquetColumnarRowInputFormat - カラムナ入力フォーマット

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

```
SerializationSchema (Python)
    │
    └─ SimpleStringSchema(charset)
           └─ gateway.jvm.SimpleStringSchema(j_char_set)

JsonRowDeserializationSchema (Python)
    │
    ├─ builder()
    │      └─ Builder()
    │
    └─ Builder
           │
           ├─ type_info(type_info)
           │      └─ self._type_info = type_info
           │
           ├─ fail_on_missing_field()
           │      └─ self._fail_on_missing_field = True
           │
           └─ build()
                  └─ JBuilder(type_info.get_java_type_info())
                  └─ j_builder.failOnMissingField()
                  └─ j_builder.build()
                  └─ JsonRowDeserializationSchema(j_schema)

CsvSchema (Python)
    │
    ├─ builder()
    │      └─ CsvSchemaBuilder()
    │
    └─ CsvSchemaBuilder
           │
           ├─ add_string_column(name)
           │      └─ j_schema_builder.addColumn(name)
           │
           └─ build()
                  └─ CsvSchema(j_schema_builder.build(), DataTypes.ROW(fields))

AvroSchema (Python)
    │
    ├─ parse_string(json_schema)
    │      └─ JSchema.Parser().parse(json_schema)
    │
    └─ AvroBulkWriters.for_generic_record(schema)
           └─ AvroWriters.forGenericRecord(schema._j_schema)
           └─ BulkWriterFactory(j_bulk_writer_factory)
```

### データフロー図

```
[Data Source]                  [PyFlink Format]              [Connector]

        │                              │                          │
        │   Binary Data (bytes)        │                          │
        └─────────────────────────────▶│                          │
                                       │                          │
                              DeserializationSchema               │
                                       │                          │
                               (JSON/CSV/Avro)                    │
                                       │                          │
                              Structured Data (Row)               │
                                       │                          │
                                       └─────────────────────────▶│
                                                                  │
                                                           DataStream
                                                                  │
                                       ◀──────────────────────────┤
                                       │                          │
                              Structured Data (Row)               │
                                       │                          │
                              SerializationSchema                 │
                                       │                          │
                               (JSON/CSV/Avro)                    │
                                       │                          │
                              Binary Data (bytes)                 │
                                       │                          │
        ◀──────────────────────────────┘                          │
        │                                                         │
 Data written to                                                  │
 External System                                                  │
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| serialization.py | `flink-python/pyflink/common/serialization.py` | ソース | シリアライズ基底 |
| json.py | `flink-python/pyflink/datastream/formats/json.py` | ソース | JSON フォーマット |
| csv.py | `flink-python/pyflink/datastream/formats/csv.py` | ソース | CSV フォーマット |
| avro.py | `flink-python/pyflink/datastream/formats/avro.py` | ソース | Avro フォーマット |
| parquet.py | `flink-python/pyflink/datastream/formats/parquet.py` | ソース | Parquet フォーマット |
| orc.py | `flink-python/pyflink/datastream/formats/orc.py` | ソース | ORC フォーマット |
| __init__.py | `flink-python/pyflink/datastream/formats/__init__.py` | ソース | モジュール初期化 |
