# 帳票設計書 3-Avro出力

## 概要

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

### 本帳票の処理概要

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

**業務上の目的・背景**：Apache Avroは、スキーマベースのデータシリアライゼーションフレームワークとして、ビッグデータエコシステムで広く利用されている。コンパクトなバイナリエンコーディング、スキーマ進化のサポート、言語中立性を特徴とし、Kafka、Hadoop、Sparkとの連携において重要な役割を果たす。Confluentスキーマレジストリとの統合により、エンタープライズ環境でのデータガバナンスにも対応する。

**帳票の利用シーン**：リアルタイムデータパイプラインにおいて、スキーマ管理が必要な場合や、後方互換性を維持しながらデータフォーマットを進化させる必要がある場合に利用される。Kafkaへのメッセージ送信、データレイクへのファイル保存、マイクロサービス間のデータ交換などのシーンで活用される。

**主要な出力内容**：
1. RowData形式の各フィールドをAvroスキーマに従ってシリアライズしたバイナリ/JSONデータ
2. スキーマ付きGenericRecordへの変換
3. Binary/JSONの2種類のエンコーディングモード
4. レガシータイムスタンプマッピングのサポート

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

**帳票の利用者**：データエンジニア、データアーキテクト、データプラットフォーム担当者

## 帳票種別

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

## 利用画面

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

## 出力形式

### 基本仕様

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

### Avro固有設定

| 項目 | 内容 | デフォルト値 |
|-----|------|-------------|
| エンコーディング | encoding | BINARY |
| 圧縮コーデック | codec | snappy |
| レガシータイムスタンプマッピング | timestamp_mapping.legacy | true |

### エンコーディングモード

| モード | 説明 |
|-------|------|
| BINARY | コンパクトで効率的なバイナリエンコーディング |
| JSON | 人間が読みやすいJSONエンコーディング |

## 帳票レイアウト

### レイアウト概要

Avro GenericRecordとして各レコードをシリアライズする。スキーマに基づいた構造化データとなる。

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

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

| データ型 | Avro型 | 変換処理 |
|---------|--------|---------|
| BOOLEAN | boolean | そのまま |
| TINYINT | int | intValue()へ変換 |
| SMALLINT | int | intValue()へ変換 |
| INTEGER | int | そのまま |
| BIGINT | long | そのまま |
| FLOAT | float | そのまま |
| DOUBLE | double | そのまま |
| CHAR/VARCHAR | Utf8 | Utf8オブジェクトに変換 |
| BINARY/VARBINARY | bytes | ByteBufferに変換 |
| DATE | int | そのまま |
| TIME | int | そのまま |
| TIMESTAMP | long | toEpochMilli()（レガシー/非レガシーで動作異なる） |
| TIMESTAMP_LTZ | long | toInstant().toEpochMilli()（非レガシーのみ） |
| DECIMAL | bytes | toUnscaledBytes()をByteBufferに変換 |
| ARRAY | array | List<Object>に変換 |
| MAP | map | Map<String, Object>に変換 |
| ROW | record | GenericData.Recordに変換 |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| Sinkテーブル定義 | CREATE TABLE文でformat='avro'を指定 | Yes |
| データ型互換性 | 出力対象がサポート型であること | Yes |
| タイムスタンプ型制約 | TIMESTAMP_LTZはレガシーモードでサポート外 | 条件付き |

### ソート順

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

### 改ページ条件

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

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

### 参照テーブル一覧

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

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

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| TINYINT変換 | Byte.intValue() | N/A | Avro intへ変換 |
| SMALLINT変換 | Short.intValue() | N/A | Avro intへ変換 |
| TIMESTAMP変換 | toInstant().toEpochMilli() | N/A | ミリ秒エポック時間 |
| DECIMAL変換 | toUnscaledBytes() | N/A | バイト配列化 |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[RowData入力] --> B[AvroRowDataSerializationSchema.serialize]
    B --> C[RowDataToAvroConverter.convert]
    C --> D[GenericRecord構築]
    D --> E[AvroSerializationSchema.serialize]
    E --> F{エンコーディング}
    F -->|BINARY| G[BinaryEncoder]
    F -->|JSON| H[JsonEncoder]
    G --> I[byte[]出力]
    H --> I
    I --> J[Sink書き込み]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| サポート外型 | RAW型の出力 | Unsupported type: {type} | サポートされる型に変換 |
| TIMESTAMP_LTZ（レガシー） | レガシーモードでTIMESTAMP_LTZ使用 | Unsupported type: {type} | レガシーモード無効化 |
| シリアライズ失敗 | フィールド変換エラー | Fail to serialize at field: {name} | データ形式の確認 |
| 行シリアライズ失敗 | 行全体の変換エラー | Failed to serialize row. | 入力データ確認 |
| Nullableスキーマエラー | 不正なUNIONスキーマ | The Avro schema is not a nullable type | スキーマ定義確認 |

## パフォーマンス要件

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

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

- Avroバイナリデータは暗号化されていないため、機密データを含む場合は通信経路の暗号化が必要
- スキーマレジストリとの通信時はSSL/TLS設定を推奨
- ファイル出力時はファイルシステムのアクセス制御に従う

## 備考

- Apache Avroライブラリを使用してシリアライズを実行
- Flink 1.19以前のレガシータイムスタンプマッピングでは、TIMESTAMP/TIMESTAMP_LTZ両方がAvro TIMESTAMPにマップされていた（非推奨）
- 正しいマッピング：TIMESTAMP→Avro LOCAL TIMESTAMP、TIMESTAMP_LTZ→Avro TIMESTAMP
- SNAPPY圧縮がデフォルトで使用される

---

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

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

### 推奨読解順序

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

まず、Avro出力で使用するオプション設定とエンコーディングモードを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | AvroFormatOptions.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFormatOptions.java` | Avroフォーマットの設定オプション（行32-87）：エンコーディング、圧縮コーデック、レガシータイムスタンプ設定 |

**読解のコツ**: AvroEncoding enum（行50-71）でBINARY/JSONの2モードを確認。legacyTimestampMapping（行73-84）はFlink 1.19以前との互換性オプション。

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

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

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

**主要処理フロー**:
1. **行89-111**: createEncodingFormat()でSerializationSchemaを生成
2. **行102-103**: AvroRowDataSerializationSchemaのインスタンス化（encoding, legacyTimestampMapping引数）
3. **行94-95**: オプション値の取得

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

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

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

**主要処理フロー**:
- **行116-124**: serialize()でRowDataをGenericRecord経由でbyte[]に変換
- **行78-254**: createConverter()で型ごとの変換ロジック
- **行256-297**: createRowConverter()で行全体の変換器を生成

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

```
AvroFormatFactory
    │
    └─ createEncodingFormat()
           └─ AvroRowDataSerializationSchema
                   │
                   ├─ RowDataToAvroConverters.createConverter()
                   │      └─ createConverter() [型別]
                   │              ├─ BOOLEAN/INTEGER/BIGINT → そのまま
                   │              ├─ TINYINT/SMALLINT → intValue()
                   │              ├─ VARCHAR → Utf8
                   │              ├─ BINARY → ByteBuffer
                   │              ├─ TIMESTAMP → toEpochMilli()
                   │              ├─ DECIMAL → ByteBuffer(toUnscaledBytes())
                   │              ├─ ARRAY → createArrayConverter()
                   │              ├─ MAP → createMapConverter()
                   │              └─ ROW → createRowConverter()
                   │
                   ├─ AvroSchemaConverter.convertToSchema()
                   │
                   └─ AvroSerializationSchema.serialize()
                          └─ GenericDatumWriter
                                 └─ Encoder (Binary/Json)
```

### データフロー図

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

RowData ───▶ RowDataToAvroConverters ───▶ GenericRecord
                      │
                      ▼
              AvroRowDataSerializationSchema.serialize()
                      │
                      ▼
              AvroSerializationSchema.serialize()
                      │
                      ▼
              GenericDatumWriter + Encoder
                      │
                      ▼
                   byte[] ───▶ Kafka/ファイル/その他Sink
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| AvroFormatFactory.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFormatFactory.java` | ソース | Table/SQL APIファクトリ |
| AvroRowDataSerializationSchema.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataSerializationSchema.java` | ソース | シリアライゼーションスキーマ |
| RowDataToAvroConverters.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java` | ソース | RowData→Avro変換 |
| AvroFormatOptions.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFormatOptions.java` | ソース | 設定オプション定義 |
| AvroSerializationSchema.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroSerializationSchema.java` | ソース | GenericRecordシリアライズ |
| AvroSchemaConverter.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java` | ソース | スキーマ変換 |
| AvroFileFormatFactory.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFileFormatFactory.java` | ソース | ファイルシステムファクトリ |
| AvroRowDataDeserializationSchema.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataDeserializationSchema.java` | ソース | デシリアライゼーション（参考） |
