# 機能設計書 42-JSON

## 概要

本ドキュメントは、Apache Flink の flink-json モジュールが提供する JSON シリアライゼーション/デシリアライゼーション機能の設計を記述する。

### 本機能の処理概要

JSON フォーマット機能は、Flink の Table API/SQL および DataStream API で使用される JSON 形式のデータを読み書きするためのシリアライザとデシリアライザを提供する。Jackson ライブラリを基盤とし、高性能な JSON 処理を実現する。

**業務上の目的・背景**：現代のデータパイプラインにおいて、JSON は最も広く使用されるデータ交換フォーマットの一つである。REST API、メッセージキュー、ログファイル等、多くのシステムが JSON を採用している。Flink で JSON データを効率的に処理することで、様々なデータソースとの統合が可能になり、リアルタイム分析やETL処理を実現できる。

**機能の利用シーン**：
- Kafka からの JSON メッセージの消費・生成
- ファイルシステムへの JSON 形式でのデータ出力
- REST API からの JSON レスポンスの処理
- CDC（Change Data Capture）ツールからの JSON 形式変更データの処理
- Table API/SQL でのJSON形式テーブルの読み書き

**主要な処理内容**：
1. JsonRowDataDeserializationSchema: JSON バイト列から RowData への変換
2. JsonRowDataSerializationSchema: RowData から JSON バイト列への変換
3. JsonParserRowDataDeserializationSchema: JsonParser ベースの高性能デシリアライザ
4. JsonFormatFactory: Table API 向けフォーマットファクトリ
5. CDC フォーマットサポート: Debezium, Canal, Maxwell, Ogg 形式対応

**関連システム・外部連携**：
- Kafka コネクタ（JSON メッセージの送受信）
- ファイルシステムコネクタ（JSON ファイルの読み書き）
- CDC ツール（Debezium, Canal, Maxwell, Oracle GoldenGate）

**権限による制御**：本機能はデータフォーマット処理のため、権限制御は利用するコネクタに委譲される。

## 関連画面

本機能はバックエンドのデータフォーマット処理機能であり、直接関連する画面はない。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | - |

## 機能種別

- データ変換処理
- シリアライゼーション/デシリアライゼーション

## 入力仕様

### 入力パラメータ

#### デシリアライゼーション設定

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| fail-on-missing-field | boolean | No | 欠落フィールドでエラーとするか | デフォルト: false |
| ignore-parse-errors | boolean | No | パースエラーを無視するか | デフォルト: false |
| timestamp-format.standard | String | No | タイムスタンプフォーマット | SQL または ISO-8601 |
| decode.json-parser.enabled | boolean | No | JsonParser 使用有無 | デフォルト: true |

#### シリアライゼーション設定

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| map-null-key.mode | String | No | null キーの処理モード | FAIL/DROP/LITERAL |
| map-null-key.literal | String | No | null キーのリテラル値 | デフォルト: "null" |
| encode.decimal-as-plain-number | boolean | No | 10進数を通常数値でエンコード | デフォルト: false |
| encode.ignore-null-fields | boolean | No | null フィールドを無視 | デフォルト: false |
| timestamp-format.standard | String | No | タイムスタンプフォーマット | SQL または ISO-8601 |

### 入力データソース

- デシリアライゼーション: JSON 形式の byte[] データ
- シリアライゼーション: Flink の RowData 内部データ構造

## 出力仕様

### 出力データ

#### デシリアライゼーション出力

| 項目名 | 型 | 説明 |
|--------|-----|------|
| rowData | RowData | Flink 内部データ形式に変換されたデータ |

#### シリアライゼーション出力

| 項目名 | 型 | 説明 |
|--------|-----|------|
| jsonBytes | byte[] | JSON 形式にシリアライズされたバイト列 |

### 出力先

- デシリアライゼーション: Flink の下流オペレーター
- シリアライゼーション: 外部システム（Kafka、ファイルシステム等）

## 処理フロー

### 処理シーケンス

#### デシリアライゼーション処理

```
1. byte[] メッセージ受信
   └─ null チェック

2. JSON パース
   └─ ObjectMapper.readTree() または JsonParser で解析

3. JsonNode から RowData への変換
   └─ JsonToRowDataConverter で型変換
   └─ 配列の場合は各要素を個別処理

4. エラーハンドリング
   └─ ignoreParseErrors=true なら継続
   └─ ignoreParseErrors=false なら IOException
```

#### シリアライゼーション処理

```
1. RowData 受信
   └─ ObjectNode の作成/再利用

2. RowData から JsonNode への変換
   └─ RowDataToJsonConverter で型変換
   └─ null フィールドの処理

3. JSON バイト列への変換
   └─ ObjectMapper.writeValueAsBytes()

4. 出力
   └─ byte[] を返却
```

### フローチャート

```mermaid
flowchart TD
    subgraph Deserialization
        A[byte[] 入力] --> B{null?}
        B -->|Yes| C[return]
        B -->|No| D[JSON パース]
        D --> E{配列?}
        E -->|Yes| F[各要素を処理]
        E -->|No| G[単一オブジェクト処理]
        F --> H[JsonToRowDataConverter]
        G --> H
        H --> I[RowData 出力]
    end

    subgraph Serialization
        J[RowData 入力] --> K[ObjectNode 作成]
        K --> L[RowDataToJsonConverter]
        L --> M[JSON バイト変換]
        M --> N[byte[] 出力]
    end
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-42-01 | 欠落フィールド処理 | fail-on-missing-field=true で欠落フィールドはエラー | デシリアライゼーション時 |
| BR-42-02 | パースエラー処理 | ignore-parse-errors=true でエラー行をスキップ | デシリアライゼーション時 |
| BR-42-03 | null キー処理 | map-null-key.mode で FAIL/DROP/LITERAL を選択 | シリアライゼーション時 |
| BR-42-04 | タイムスタンプ形式 | SQL: "yyyy-MM-dd HH:mm:ss.s", ISO-8601: "yyyy-MM-ddTHH:mm:ss.s" | 両方向 |
| BR-42-05 | 配列入力処理 | 配列の場合は各要素を個別のレコードとして出力 | デシリアライゼーション時 |
| BR-42-06 | null フィールド処理 | encode.ignore-null-fields=true で null フィールドを出力しない | シリアライゼーション時 |

### 計算ロジック

**タイムスタンプ変換**:
- SQL 形式: `yyyy-MM-dd HH:mm:ss.s{precision}` (精度は0-9)
- ISO-8601 形式: `yyyy-MM-ddTHH:mm:ss.s{precision}`

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

本機能はデータフォーマット変換のため、直接的なデータベース操作は行わない。

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データフォーマット機能のためDB操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | パースエラー | JSON 構文エラー | ignore-parse-errors 設定確認 |
| RuntimeException | シリアライゼーションエラー | RowData 変換失敗 | スキーマ定義確認 |
| IllegalArgumentException | 設定エラー | 無効な設定値 | 設定オプション確認 |

### リトライ仕様

- ignore-parse-errors=true の場合、パースエラーのレコードをスキップして処理継続
- シリアライゼーションエラーはリトライなし（RuntimeException）

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

本機能はステートレスなデータ変換のため、トランザクション管理は不要。

## パフォーマンス要件

- JsonParser 使用時は ObjectMapper より高性能
- ObjectNode の再利用によりメモリ割り当て削減（ignoreNullFields=false の場合）
- Jackson の BigDecimal 設定で精度を保持

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

- 入力 JSON の検証は設定に依存
- 大きな JSON オブジェクトによるメモリ消費に注意

## 備考

- CDC フォーマット（Debezium, Canal, Maxwell, Ogg）は個別のサブパッケージで提供
- JsonParser 使用時はネストしたプロジェクションをサポート

---

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

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

### 推奨読解順序

#### Step 1: データ構造と設定オプションを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | JsonFormatOptions.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatOptions.java` | 利用可能な設定オプションの定義 |

**読解のコツ**: ConfigOption の定義パターンを理解する。defaultValue() と withDescription() で設定の意味を把握できる。

#### Step 2: ファクトリクラスを理解する（エントリーポイント）

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | JsonFormatFactory.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java` | Table API からのフォーマット生成 |

**主要処理フロー**:
1. **63行目**: IDENTIFIER = "json" でフォーマット識別子を定義
2. **66-116行目**: createDecodingFormat() でデシリアライザ生成
3. **139-172行目**: createEncodingFormat() でシリアライザ生成
4. **185-196行目**: optionalOptions() で設定オプションを列挙

#### Step 3: シリアライゼーション処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | JsonRowDataSerializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java` | RowData から JSON への変換 |
| 3-2 | RowDataToJsonConverters.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java` | 型別変換ロジック |

**主要処理フロー（JsonRowDataSerializationSchema）**:
- **76-96行目**: コンストラクタで RowDataToJsonConverter を初期化
- **99-107行目**: open() で ObjectMapper を設定
- **110-121行目**: serialize() で RowData を byte[] に変換

#### Step 4: デシリアライゼーション処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | JsonRowDataDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java` | JSON から RowData への変換 |
| 4-2 | AbstractJsonDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java` | デシリアライザ共通基底クラス |
| 4-3 | JsonToRowDataConverters.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java` | 型別変換ロジック |

**主要処理フロー（JsonRowDataDeserializationSchema）**:
- **55-65行目**: コンストラクタで JsonToRowDataConverter を初期化
- **68-101行目**: deserialize() で byte[] を RowData に変換
- **74-88行目**: 配列入力の場合は各要素を個別処理

#### Step 5: 高性能 JsonParser 実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | JsonParserRowDataDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserRowDataDeserializationSchema.java` | JsonParser ベースの実装 |
| 5-2 | JsonParserToRowDataConverters.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserToRowDataConverters.java` | JsonParser 用変換ロジック |

#### Step 6: CDC フォーマットを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 6-1 | DebeziumJsonDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java` | Debezium CDC 形式 |
| 6-2 | CanalJsonDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java` | Canal CDC 形式 |
| 6-3 | MaxwellJsonDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonDeserializationSchema.java` | Maxwell CDC 形式 |

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

```
JsonFormatFactory
    │
    ├─ createDecodingFormat()
    │      ├─ JsonParserRowDataDeserializationSchema (json-parser enabled)
    │      │      └─ JsonParserToRowDataConverters
    │      │
    │      └─ JsonRowDataDeserializationSchema (json-parser disabled)
    │             └─ JsonToRowDataConverters
    │
    └─ createEncodingFormat()
           └─ JsonRowDataSerializationSchema
                  └─ RowDataToJsonConverters

CDC Formats (Debezium/Canal/Maxwell/Ogg)
    └─ 各 FormatFactory
           ├─ DeserializationSchema
           └─ SerializationSchema
```

### データフロー図

```
[デシリアライゼーション]
byte[] (JSON) ───▶ JsonNode/JsonParser ───▶ Converter ───▶ RowData

[シリアライゼーション]
RowData ───▶ Converter ───▶ ObjectNode ───▶ byte[] (JSON)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| JsonFormatFactory.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java` | ソース | Table API ファクトリ |
| JsonFormatOptions.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatOptions.java` | ソース | 設定オプション定義 |
| JsonRowDataSerializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java` | ソース | シリアライザ |
| JsonRowDataDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java` | ソース | デシリアライザ |
| AbstractJsonDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java` | ソース | デシリアライザ基底クラス |
| JsonParserRowDataDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserRowDataDeserializationSchema.java` | ソース | 高性能デシリアライザ |
| JsonToRowDataConverters.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java` | ソース | JSON→RowData 変換 |
| RowDataToJsonConverters.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java` | ソース | RowData→JSON 変換 |
| JsonSerializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSerializationSchema.java` | ソース | DataStream API シリアライザ |
| JsonDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonDeserializationSchema.java` | ソース | DataStream API デシリアライザ |
| DebeziumJsonFormatFactory.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonFormatFactory.java` | ソース | Debezium CDC ファクトリ |
| CanalJsonFormatFactory.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonFormatFactory.java` | ソース | Canal CDC ファクトリ |
| MaxwellJsonFormatFactory.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonFormatFactory.java` | ソース | Maxwell CDC ファクトリ |
| OggJsonFormatFactory.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonFormatFactory.java` | ソース | Ogg CDC ファクトリ |
