# 機能設計書 43-Avro

## 概要

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

### 本機能の処理概要

Avro フォーマット機能は、Apache Avro 形式のデータを読み書きするためのシリアライザとデシリアライザを提供する。Avro はスキーマ駆動のバイナリシリアライゼーションフォーマットであり、データ圧縮とスキーマ進化をサポートする。

**業務上の目的・背景**：Apache Avro は Hadoop エコシステムで広く採用されているデータシリアライゼーションフォーマットである。スキーマ駆動でありながらコンパクトなバイナリ形式を提供し、スキーマ進化（互換性のあるスキーマ変更）をサポートする。Flink で Avro を扱うことで、既存の Hadoop/Kafka エコシステムとのシームレスな統合が可能になる。

**機能の利用シーン**：
- Kafka からの Avro メッセージの消費・生成
- Avro 形式のファイル読み書き
- Schema Registry と連携した Avro データ処理
- Hadoop エコシステムとのデータ連携
- スキーマ進化が必要なデータパイプライン

**主要な処理内容**：
1. AvroRowDataDeserializationSchema: Avro バイト列から RowData への変換
2. AvroRowDataSerializationSchema: RowData から Avro バイト列への変換
3. AvroSchemaConverter: Flink 型と Avro スキーマの相互変換
4. AvroSerializer: Flink 状態管理用 Avro シリアライザ
5. バルクフォーマット: Avro ファイルの一括読み書き

**関連システム・外部連携**：
- Kafka コネクタ（Avro メッセージの送受信）
- ファイルシステムコネクタ（Avro ファイルの読み書き）
- Confluent Schema Registry（別モジュール flink-avro-confluent-registry）

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

## 関連画面

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

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

## 機能種別

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

## 入力仕様

### 入力パラメータ

#### 共通設定

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| encoding | AvroEncoding | No | エンコーディング方式 | BINARY または JSON、デフォルト: BINARY |
| timestamp_mapping.legacy | boolean | No | レガシータイムスタンプマッピング使用 | デフォルト: true |

#### ファイル出力設定

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| codec | String | No | 圧縮コーデック | デフォルト: snappy |

### 入力データソース

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

## 出力仕様

### 出力データ

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

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

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

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

### 出力先

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

## 処理フロー

### 処理シーケンス

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

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

2. Avro GenericRecord への変換
   └─ nestedSchema.deserialize() で Avro デコード

3. GenericRecord から RowData への変換
   └─ AvroToRowDataConverter で型変換

4. エラーハンドリング
   └─ IOException をスロー
```

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

```
1. RowData 受信
   └─ Avro Schema の取得

2. RowData から GenericRecord への変換
   └─ RowDataToAvroConverter で型変換

3. GenericRecord から byte[] への変換
   └─ nestedSchema.serialize() で Avro エンコード

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

### フローチャート

```mermaid
flowchart TD
    subgraph Deserialization
        A[byte[] 入力] --> B{null?}
        B -->|Yes| C[null 返却]
        B -->|No| D[Avro デコード]
        D --> E[GenericRecord]
        E --> F[AvroToRowDataConverter]
        F --> G[RowData 出力]
    end

    subgraph Serialization
        H[RowData 入力] --> I[Schema 取得]
        I --> J[RowDataToAvroConverter]
        J --> K[GenericRecord]
        K --> L[Avro エンコード]
        L --> M[byte[] 出力]
    end
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-43-01 | エンコーディング選択 | BINARY は高効率、JSON は可読性重視 | 全処理 |
| BR-43-02 | タイムスタンプマッピング | legacy=true で 1.19以前の互換動作 | タイムスタンプ型処理時 |
| BR-43-03 | 圧縮コーデック | snappy がデフォルト、null/deflate 等も選択可 | ファイル出力時 |
| BR-43-04 | スキーマ変換 | Flink RowType を Avro Schema に自動変換 | シリアライゼーション初期化時 |
| BR-43-05 | null 入力処理 | null バイト列は null を返却 | デシリアライゼーション時 |

### 計算ロジック

**タイムスタンプマッピング**:
- legacy=true: TIMESTAMP と TIMESTAMP_LTZ の両方を Avro TIMESTAMP にマップ（1.19以前互換）
- legacy=false: TIMESTAMP を Avro LOCAL TIMESTAMP に、TIMESTAMP_LTZ を Avro TIMESTAMP にマップ（正しいマッピング）

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

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

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | デシリアライゼーションエラー | Avro 形式不正 | 入力データ形式確認 |
| RuntimeException | シリアライゼーションエラー | RowData 変換失敗 | スキーマ定義確認 |

### リトライ仕様

- エラー発生時はリトライなし（例外をスロー）

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

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

## パフォーマンス要件

- Binary エンコーディングは JSON より高効率
- Snappy 圧縮でファイルサイズと処理速度のバランスを確保
- GenericRecord を介した2段階変換

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

- スキーマ検証は Avro ライブラリに委譲
- 大きな Avro オブジェクトによるメモリ消費に注意

## 備考

- Schema Registry 連携は別モジュール flink-avro-confluent-registry で提供
- JodaTime 依存が必要な場合あり（日付/時刻の論理型使用時）

---

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

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

### 推奨読解順序

#### Step 1: 設定オプションを理解する

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

**読解のコツ**: AvroEncoding enum の BINARY/JSON の違いを理解する。timestamp_mapping.legacy の背景（1.19以前の互換性問題）を把握する。

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

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

**主要処理フロー**:
1. **57行目**: IDENTIFIER = "avro" でフォーマット識別子を定義
2. **60-87行目**: createDecodingFormat() でデシリアライザ生成
3. **90-111行目**: createEncodingFormat() でシリアライザ生成
4. **124-129行目**: optionalOptions() で設定オプションを列挙

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

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

**主要処理フロー（AvroRowDataSerializationSchema）**:
- **58-60行目**: デフォルトコンストラクタ（BINARY エンコーディング）
- **82-90行目**: legacyTimestampMapping 対応コンストラクタ
- **106-113行目**: open() で Schema を取得
- **116-124行目**: serialize() で RowData を byte[] に変換

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

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

**主要処理フロー（AvroRowDataDeserializationSchema）**:
- **68-70行目**: デフォルトコンストラクタ
- **98-109行目**: legacyTimestampMapping 対応コンストラクタ
- **130-132行目**: open() で nestedSchema を初期化
- **135-145行目**: deserialize() で byte[] を RowData に変換

#### Step 5: スキーマ変換を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | AvroSchemaConverter.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java` | Flink 型と Avro Schema の変換 |

#### Step 6: バルクフォーマットを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 6-1 | AvroWriters.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroWriters.java` | Avro ファイル書き込み |
| 6-2 | AbstractAvroBulkFormat.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AbstractAvroBulkFormat.java` | Avro ファイル読み込み基盤 |

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

```
AvroFormatFactory
    │
    ├─ createDecodingFormat()
    │      └─ AvroRowDataDeserializationSchema
    │             ├─ AvroDeserializationSchema (nestedSchema)
    │             └─ AvroToRowDataConverters
    │
    └─ createEncodingFormat()
           └─ AvroRowDataSerializationSchema
                  ├─ AvroSerializationSchema (nestedSchema)
                  └─ RowDataToAvroConverters

AvroSchemaConverter
    └─ convertToSchema() / convertToDataType()
           └─ Flink RowType <-> Avro Schema
```

### データフロー図

```
[デシリアライゼーション]
byte[] (Avro) ───▶ GenericRecord ───▶ AvroToRowDataConverter ───▶ RowData

[シリアライゼーション]
RowData ───▶ RowDataToAvroConverter ───▶ GenericRecord ───▶ byte[] (Avro)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| AvroFormatFactory.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFormatFactory.java` | ソース | Table API ファクトリ |
| AvroFormatOptions.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFormatOptions.java` | ソース | 設定オプション定義 |
| AvroRowDataSerializationSchema.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataSerializationSchema.java` | ソース | シリアライザ |
| AvroRowDataDeserializationSchema.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataDeserializationSchema.java` | ソース | デシリアライザ |
| AvroSerializationSchema.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroSerializationSchema.java` | ソース | Avro エンコード |
| AvroDeserializationSchema.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java` | ソース | Avro デコード |
| RowDataToAvroConverters.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java` | ソース | RowData→Avro 変換 |
| AvroToRowDataConverters.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java` | ソース | Avro→RowData 変換 |
| AvroSchemaConverter.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java` | ソース | スキーマ変換 |
| AvroSerializer.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java` | ソース | Flink 状態用シリアライザ |
| AvroWriters.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroWriters.java` | ソース | バルク書き込み |
| AbstractAvroBulkFormat.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AbstractAvroBulkFormat.java` | ソース | バルク読み込み基盤 |
| AvroFileFormatFactory.java | `flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFileFormatFactory.java` | ソース | ファイルフォーマットファクトリ |
