# 帳票設計書 2-JSON出力

## 概要

本ドキュメントは、Apache Flink のJSON出力機能に関する帳票設計書である。JsonRowDataSerializationSchemaを中心としたRowDataのJSONシリアライズ機能について、その仕様と実装詳細を記載する。

### 本帳票の処理概要

本帳票は、Apache Flink のストリーミング処理またはバッチ処理において、RowData形式のデータをJSON（JavaScript Object Notation）フォーマットでシリアライズし、外部システムやファイルシステムに出力する機能を提供する。

**業務上の目的・背景**：JSONはWebアプリケーションやREST API、NoSQLデータベースとのデータ交換において最も広く利用されるフォーマットである。Flinkのストリーミング処理結果をJSON形式で出力することで、Kafka、Elasticsearch、MongoDB等のシステムとの連携が可能となる。また、人間が読みやすい形式でログやデバッグ情報を出力する目的でも利用される。

**帳票の利用シーン**：リアルタイムデータ処理パイプラインにおいて、処理結果をKafkaトピックにJSON形式で書き込む場合や、REST APIへのデータ送信、Elasticsearchへのインデックス作成などで利用される。マイクロサービス間のデータ連携、ログ分析基盤へのデータ投入、イベントソーシングアーキテクチャでのイベント出力などのシーンで活用される。

**主要な出力内容**：
1. RowData形式の各フィールドをJSONオブジェクトとしてシリアライズしたデータ
2. 設定可能なタイムスタンプフォーマット（SQL/ISO-8601）による日時表現
3. Map型のNullキー処理モード（FAIL/DROP/LITERAL）
4. Decimal型の表記制御、Nullフィールドの除外オプション

**帳票の出力タイミング**：Flink Table/SQL APIにおいてSinkとして設定されたタイミングでストリーミングデータが継続的にシリアライズされる。Kafkaへの出力やファイルシステムへの書き込み時にJSONバイト列として出力される。

**帳票の利用者**：データエンジニア、バックエンドエンジニア、データ連携担当者

## 帳票種別

データフォーマット出力（ストリーミング/バッチ JSON形式）

## 利用画面

| 画面No | 画面名 | URL/ルーティング | 出力操作 |
|--------|--------|-----------------|---------|
| - | Flink SQL CLI | - | CREATE TABLE ... WITH ('format' = 'json') |
| - | Flink Table API | - | tableEnv.connect().withFormat(new Json()) |
| - | Kafka Connector | - | 'value.format' = 'json' |

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | JSON（JavaScript Object Notation） |
| 用紙サイズ | N/A（データフォーマット） |
| 向き | N/A |
| ファイル名 | Sinkの設定による |
| 出力方法 | Kafka/ファイルシステム/その他Connector |
| 文字コード | UTF-8 |

### JSON固有設定

| 項目 | 内容 | デフォルト値 |
|-----|------|-------------|
| タイムスタンプフォーマット | timestamp-format.standard | SQL |
| Map Nullキーモード | map-null-key.mode | FAIL |
| Map Nullキーリテラル | map-null-key.literal | null |
| Decimal平文表記 | encode.decimal-as-plain-number | false |
| Nullフィールド無視 | encode.ignore-null-fields | false |
| 欠損フィールドでエラー | fail-on-missing-field | false |
| パースエラー無視 | ignore-parse-errors | false |
| JSONParser有効 | decode.json-parser.enabled | true |

## 帳票レイアウト

### レイアウト概要

JSONオブジェクトとして各レコードをシリアライズする。

```
{
  "field1": value1,
  "field2": value2,
  ...
  "fieldN": valueN
}
```

### フィールド型変換仕様

| データ型 | JSON表現 | 変換処理 |
|---------|--------|---------|
| BOOLEAN | true/false | booleanNode |
| TINYINT | 数値 | numberNode |
| SMALLINT | 数値 | numberNode |
| INTEGER | 数値 | numberNode |
| BIGINT | 数値 | numberNode |
| FLOAT | 数値 | numberNode |
| DOUBLE | 数値 | numberNode |
| CHAR/VARCHAR | "文字列" | textNode |
| BINARY/VARBINARY | バイナリ | binaryNode |
| DATE | "yyyy-MM-dd" | ISO_LOCAL_DATE |
| TIME | "HH:mm:ss" | SQL_TIME_FORMAT |
| TIMESTAMP | "yyyy-MM-dd HH:mm:ss"または"yyyy-MM-ddTHH:mm:ss" | タイムスタンプフォーマット設定依存 |
| TIMESTAMP_LTZ | タイムゾーン付き形式 | タイムスタンプフォーマット設定依存 |
| DECIMAL | 数値（科学表記制御可） | numberNode |
| ARRAY | [要素, 要素, ...] | ArrayNode |
| MAP | {"key": value, ...} | ObjectNode（文字列キーのみ） |
| ROW | {...} | ObjectNode |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| Sinkテーブル定義 | CREATE TABLE文でformat='json'を指定 | Yes |
| データ型互換性 | 出力対象がサポート型であること | Yes |
| Mapキー型 | MAP型のキーは文字列型であること | Yes（MAP使用時） |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| 1 | データ到着順（ストリーミング）/ 入力順（バッチ） | N/A |

### 改ページ条件

N/A（レコード単位のシリアライズ）

## データベース参照仕様

### 参照テーブル一覧

本機能はFlinkのSink機能であり、上流からのデータストリームまたはテーブルをそのまま出力する。

| テーブル名 | 用途 | 結合条件 |
|-----------|------|---------|
| 上流テーブル/ストリーム | 出力対象データ | N/A |

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| BigDecimal変換 | stripTrailingZeros() or そのまま | 設定依存 | encode.decimal-as-plain-number |
| タイムスタンプ変換 | toLocalDateTime() / toInstant() | N/A | timestamp-format.standard |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[RowData入力] --> B[JsonRowDataSerializationSchema.serialize]
    B --> C[RowDataToJsonConverter.convert]
    C --> D[ObjectNode構築]
    D --> E[ObjectMapper.writeValueAsBytes]
    E --> F[byte[]出力]
    F --> G[Sink書き込み]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| サポート外型 | RAW型の出力 | Not support to parse type: {type} | サポートされる型に変換 |
| Map非文字列キー | MAPのキーが文字列以外 | JSON format doesn't support non-string as key type of map | キー型を文字列に変更 |
| Map Nullキー（FAILモード） | MAPキーがnullでFAILモード | JSON format doesn't support to serialize map data with null keys | map-null-key.mode設定変更 |
| シリアライズ失敗 | フィールド変換エラー | Fail to serialize at field: {name} | データ形式の確認 |
| 行シリアライズ失敗 | 行全体の変換エラー | Could not serialize row '{row}'. | 入力データ確認 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | 無制限（ストリーミング） |
| 目標出力時間 | リアルタイム処理対応 |
| 同時出力数上限 | 並列度に依存 |

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

- JSONデータは平文テキストのため、機密データを含む場合は通信経路の暗号化が必要
- Kafkaへの出力時はSSL/TLS設定を推奨
- ファイル出力時はファイルシステムのアクセス制御に従う

## 備考

- Jackson ObjectMapper（jackson-databind）を使用してJSON変換を実行
- ignoreNullFieldsがtrueの場合、各serialize呼び出しで新しいObjectNodeを作成
- MAP型はキーが文字列型のみサポート（非文字列キーはUnsupportedOperationException）

---

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

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

### 推奨読解順序

#### Step 1: データ構造を理解する

まず、JSON出力で使用するオプション設定を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | JsonFormatOptions.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatOptions.java` | JSONフォーマットの設定オプション（行27-102）：タイムスタンプフォーマット、Mapキー処理、Decimal表記等 |

**読解のコツ**: MapNullKeyModeのenum（行95-99）で、Map型のnullキー処理の3つのモード（FAIL/DROP/LITERAL）を理解する。

#### Step 2: エントリーポイントを理解する

処理の起点となるFactoryクラスを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | JsonFormatFactory.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java` | Table/SQL APIのフォーマットファクトリ（行61-209）：createEncodingFormat()がSink用エンコーダを生成 |

**主要処理フロー**:
1. **行138-172**: createEncodingFormat()でSerializationSchemaを生成
2. **行158-164**: JsonRowDataSerializationSchemaのインスタンス化
3. **行144-151**: オプション値の取得と設定

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

RowDataからJSONバイト列への変換処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | JsonRowDataSerializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java` | シリアライゼーションスキーマ本体（行46-150）：serialize()がRowDataをbyte[]に変換 |
| 3-2 | RowDataToJsonConverters.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java` | 型別変換ロジック（行59-367）：各Flink型からJsonNodeへの変換 |

**主要処理フロー**:
- **行110-121**: serialize()でRowDataをJsonNode経由でbyte[]に変換
- **行100-156**: createNotNullConverter()で型ごとの変換ロジック
- **行315-356**: createRowConverter()で行全体の変換器を生成
- **行259-313**: createMapConverter()でMAP型の変換処理

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

```
JsonFormatFactory
    │
    └─ createEncodingFormat()
           └─ JsonRowDataSerializationSchema
                   │
                   ├─ RowDataToJsonConverters
                   │      └─ createConverter() [型別]
                   │              ├─ BOOLEAN → booleanNode
                   │              ├─ INTEGER → numberNode
                   │              ├─ VARCHAR → textNode
                   │              ├─ DATE → createDateConverter()
                   │              ├─ TIMESTAMP → createTimestampConverter()
                   │              ├─ DECIMAL → createDecimalConverter()
                   │              ├─ ARRAY → createArrayConverter()
                   │              ├─ MAP → createMapConverter()
                   │              └─ ROW → createRowConverter()
                   │
                   └─ ObjectMapper.writeValueAsBytes()
```

### データフロー図

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

RowData ───▶ RowDataToJsonConverters ───▶ JsonNode(ObjectNode)
                      │
                      ▼
              JsonRowDataSerializationSchema.serialize()
                      │
                      ▼
              ObjectMapper.writeValueAsBytes()
                      │
                      ▼
                   byte[] ───▶ Kafka/ファイル/その他Sink
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| JsonFormatFactory.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java` | ソース | Table/SQL APIファクトリ |
| JsonRowDataSerializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataSerializationSchema.java` | ソース | シリアライゼーションスキーマ |
| RowDataToJsonConverters.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java` | ソース | RowData→JSON変換 |
| JsonFormatOptions.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatOptions.java` | ソース | 設定オプション定義 |
| JsonFormatOptionsUtil.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatOptionsUtil.java` | ソース | オプションユーティリティ |
| JsonRowDataDeserializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java` | ソース | デシリアライゼーション（参考） |
| JsonSerializationSchema.java | `flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSerializationSchema.java` | ソース | 汎用シリアライゼーション |
