# 機能設計書 32-Avroデータフォーマット

## 概要

本ドキュメントは、Apache SparkにおけるApache Avroフォーマットのデータ読み書き機能に関する機能設計書である。Data Source V2 APIに基づくAvroデータソースプロバイダー、および`from_avro`/`to_avro`関数によるバイナリAvroデータとSpark SQL型の相互変換機構を詳細に記述する。

### 本機能の処理概要

**業務上の目的・背景**：Apache Avroは、コンパクトなバイナリフォーマットとスキーマ進化（Schema Evolution）への対応力から、Kafkaメッセージングや分散データストアで広く使われている。SparkでのネイティブなAvroサポートにより、ETLパイプラインやリアルタイムデータ処理でAvroデータを直接読み書きでき、不必要な中間変換を排除してパフォーマンスとデータ整合性を向上させる。

**機能の利用シーン**：KafkaからAvroエンコードされたメッセージを読み取りStructured Streamingで処理する場合、Avroファイルとして保存されたデータレイクの履歴データを分析する場合、既存のAvroスキーマ定義に準拠したデータ出力が求められる場合。

**主要な処理内容**：
1. Avroファイルの読み込み：Data Source V2 APIを通じてAvroフォーマットファイルを読み込み、Sparkのカラムナ形式に変換する
2. Avroファイルの書き出し：SparkのDataFrame/Datasetをavroフォーマットでファイルに書き出す
3. `from_avro`関数：バイナリAvroデータをSpark SQL構造体に変換するSQL関数
4. `to_avro`関数：Spark SQLデータをバイナリAvro形式に変換するSQL関数
5. スキーマ変換：Avroスキーマ（JSON）とSpark SQL StructTypeの相互変換
6. スキーマ進化への対応：reader schemaとwriter schemaの差異を自動解決

**関連システム・外部連携**：Apache Avroライブラリ、Confluent Schema Registry（オプション）、Kafkaコネクタ（from_avro/to_avro経由）

**権限による制御**：ファイルシステムのアクセス権限に従う。特別な権限制御はない。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能に直接関連するWeb UI画面はない |

## 機能種別

データ連携 / データ変換

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| path | String | Yes（ファイル読み書き） | Avroファイルのパス | 有効なファイルパス |
| jsonFormatSchema | String | Yes（from_avro） | Avroスキーマ（JSON文字列） | 有効なAvro Schema JSON |
| options | Map[String, String] | No | 読み書きオプション | mode, datetimeRebaseMode等 |
| data (from_avro) | Binary | Yes | Avroエンコードされたバイナリデータ | バイナリ型 |
| data (to_avro) | Any | Yes | Spark SQL型のデータ | 変換可能な型 |

### 入力データソース

- Avro形式のファイル（HDFS, S3, ローカルFS等）
- Kafkaメッセージ等のバイナリAvroデータ（from_avro関数経由）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| DataFrame | DataFrame | Avroファイル読み込み結果（Spark SQL型） |
| Binary | Array[Byte] | to_avro変換結果（Avroバイナリ） |
| StructType | StructType | from_avro変換結果（Spark SQL構造体） |

### 出力先

- Spark DataFrame（読み込み時）
- Avroファイル（書き出し時）
- Kafkaシンク等（to_avroバイナリ）

## 処理フロー

### 処理シーケンス

```
1. データ読み込み（ファイル）
   └─ AvroDataSourceV2.getTable() → AvroTable → AvroScanBuilder → AvroScan
2. パーティション読み取り
   └─ AvroPartitionReaderFactory でパーティション単位のリーダー生成
3. デシリアライゼーション
   └─ AvroDeserializer で Avro GenericRecord → InternalRow 変換
4. from_avro 関数実行
   └─ AvroDataToCatalyst.nullSafeEval() でバイナリ → InternalRow
5. to_avro 関数実行
   └─ CatalystDataToAvro.nullSafeEval() で InternalRow → バイナリ
6. スキーマ変換
   └─ SchemaConverters.toSqlType() / toAvroType() で相互変換
```

### フローチャート

```mermaid
flowchart TD
    A[Avroデータソース] --> B{操作種別}
    B -->|ファイル読み込み| C[AvroDataSourceV2]
    B -->|from_avro| D[AvroDataToCatalyst]
    B -->|to_avro| E[CatalystDataToAvro]
    B -->|ファイル書き出し| F[AvroWrite]
    C --> G[AvroPartitionReaderFactory]
    G --> H[AvroDeserializer]
    H --> I[InternalRow]
    D --> J[GenericDatumReader]
    J --> H
    E --> K[AvroSerializer]
    K --> L[GenericDatumWriter]
    L --> M[Binary Output]
    F --> K
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | ParseMode制御 | PermissiveMode（不正レコードをnull変換）とFailFastMode（例外発生）を選択可能 | from_avro実行時 |
| BR-02 | スキーマ進化 | reader schemaとwriter schemaの差異はAvroの互換性ルールに従って自動解決 | ファイル読み込み/from_avro時 |
| BR-03 | Nullable変換 | PermissiveModeではスキーマの全フィールドがnullableとなる | from_avro + PermissiveMode |
| BR-04 | 再帰フィールド制限 | recursiveFieldMaxDepthオプションで再帰的スキーマの深さを制限 | AvroDeserializer生成時 |

### 計算ロジック

- **from_avro**: BinaryDecoder → GenericDatumReader.read() → AvroDeserializer.deserialize() → InternalRow
- **to_avro**: AvroSerializer.serialize() → GenericDatumWriter.write() → BinaryEncoder → byte[]
- **スキーマ変換**: SchemaConverters が Avro の RECORD/ARRAY/MAP/UNION/ENUM/FIXED 等を StructType/ArrayType/MapType 等に変換

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし（ファイルI/O） |

### テーブル別操作詳細

データベース操作は発生しない。ファイルシステム上のAvroファイルに対するI/O操作のみ。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| MALFORMED_RECORD | SparkException | FailFastModeで不正なAvroレコードを検出 | データ修正またはPermissiveModeに変更 |
| PARSE_MODE_UNSUPPORTED | QueryCompilationError | Permissive/FailFast以外のモード指定 | 有効なモードを指定 |
| SCHEMA_MISMATCH | RuntimeException | reader/writerスキーマの互換性なし | スキーマの互換性を確認 |

### リトライ仕様

Avroファイル読み書きはSpark標準のタスクリトライメカニズムに従う。

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

ファイル書き出しはSparkの標準ファイルコミットプロトコルに従い、投機的実行やタスク失敗時の部分ファイル処理が行われる。

## パフォーマンス要件

- Avroのバイナリフォーマットはテキスト形式（CSV/JSON）より読み書き速度が高速
- スキーマがデータに同梱されるため、スキーマ推論のオーバーヘッドが小さい
- from_avro/to_avro関数はコード生成（CodeGen）をサポートし、JITコンパイル可能

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

- ファイルシステムのアクセス権限に従う
- Schema Registry連携時は認証情報の安全な管理が必要

## 備考

- Avroコネクタは `connector/avro/` 配下に独立モジュールとして配置されている
- Data Source V2 API（AvroDataSourceV2）とV1 API（AvroFileFormat）の両方に対応
- `useStableIdForUnionType` オプションでUNION型の安定したID付与が可能

---

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

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

### 推奨読解順序

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

AvroスキーマとSpark SQL型の対応関係を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SchemaConverters.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/` | toSqlType/toAvroTypeの変換ロジック |
| 1-2 | AvroOptions.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/` | 読み書きオプション定義 |

**読解のコツ**: SchemaConvertersはAvro Schema（JSONベース）とSparkのStructType間の変換を担う。UNION型がnullableフィールドに変換される点に注意。

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

Data Source V2 APIのエントリーポイントを把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AvroDataSourceV2.scala | `connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroDataSourceV2.scala` | shortName()="avro"、getTable()メソッド |
| 2-2 | AvroTable.scala | `connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala` | テーブル定義とScan/Writeビルダー |

**主要処理フロー**:
1. **26-30行目（AvroDataSourceV2.scala）**: AvroDataSourceV2がFileDataSourceV2を拡張、shortName="avro"
2. **32-37行目（AvroDataSourceV2.scala）**: getTable()でAvroTableインスタンスを生成

#### Step 3: from_avro/to_avro関数を理解する

SQL関数としてのAvro変換処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | AvroDataToCatalyst.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala` | from_avro実装 |
| 3-2 | CatalystDataToAvro.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala` | to_avro実装 |

**主要処理フロー**:
- **33-37行目（AvroDataToCatalyst.scala）**: case classの定義、child式・JSONスキーマ・オプションを保持
- **103-128行目（AvroDataToCatalyst.scala）**: `nullSafeEval` - BinaryDecoder経由でAvroデータを読み取り、AvroDeserializerで変換。ParseModeに応じてエラーハンドリング
- **30-32行目（CatalystDataToAvro.scala）**: case classの定義、child式と出力スキーマを保持
- **51-58行目（CatalystDataToAvro.scala）**: `nullSafeEval` - AvroSerializerで変換後、GenericDatumWriterでバイナリ出力

#### Step 4: パッケージ関数を理解する

ユーザー向けAPIとしてのfrom_avro/to_avro関数呼び出し口を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | package.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/package.scala` | from_avro/to_avroのColumn API |
| 4-2 | functions.scala | `sql/api/src/main/scala/org/apache/spark/sql/avro/functions.scala` | SQL API向けfrom_avro/to_avro |

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

```
spark.read.format("avro").load(path)
    │
    ├─ AvroDataSourceV2.getTable()
    │      └─ AvroTable
    │             ├─ AvroScanBuilder → AvroScan
    │             │      └─ AvroPartitionReaderFactory
    │             │             └─ AvroDeserializer
    │             └─ AvroWrite
    │                    └─ AvroSerializer
    │
from_avro(col, schema)
    └─ AvroDataToCatalyst
           ├─ GenericDatumReader.read()
           └─ AvroDeserializer.deserialize()

to_avro(col, schema)
    └─ CatalystDataToAvro
           ├─ AvroSerializer.serialize()
           └─ GenericDatumWriter.write()
```

### データフロー図

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

Avroファイル            AvroPartitionReaderFactory        DataFrame
(Binary)          ──▶  → AvroDeserializer           ──▶  (InternalRow)

DataFrame               AvroSerializer                    Avroファイル
(InternalRow)     ──▶  → GenericDatumWriter         ──▶  (Binary)

Kafkaバイナリ           AvroDataToCatalyst               Spark SQL値
(from_avro)       ──▶  → GenericDatumReader         ──▶  (InternalRow)

Spark SQL値             CatalystDataToAvro               Kafkaバイナリ
(to_avro)         ──▶  → AvroSerializer             ──▶  (byte[])
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| AvroDataSourceV2.scala | `connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/` | ソース | V2データソースエントリーポイント |
| AvroTable.scala | `connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/` | ソース | テーブル抽象化 |
| AvroScan.scala | `connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/` | ソース | スキャン定義 |
| AvroScanBuilder.scala | `connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/` | ソース | スキャンビルダー |
| AvroPartitionReaderFactory.scala | `connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/` | ソース | パーティションリーダー |
| AvroWrite.scala | `connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/` | ソース | 書き込み処理 |
| AvroDataToCatalyst.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/` | ソース | from_avro式 |
| CatalystDataToAvro.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/` | ソース | to_avro式 |
| SchemaOfAvro.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/` | ソース | スキーマ推論 |
| AvroExpressionEvalUtils.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/` | ソース | 式評価ユーティリティ |
| package.scala | `connector/avro/src/main/scala/org/apache/spark/sql/avro/` | ソース | パブリックAPI定義 |
| functions.scala | `sql/api/src/main/scala/org/apache/spark/sql/avro/` | ソース | SQL関数API |
