# 機能設計書 24-Parquetサポート

## 概要

本ドキュメントは、Apache SparkにおけるApache Parquetカラムナフォーマットの読み書き機能の設計について記述する。述語プッシュダウンやカラムプルーニングによる効率的なI/Oを実現する。

### 本機能の処理概要

SparkのParquetサポートは、Apache Parquetカラムナストレージフォーマットの読み書き機能を提供する。Parquetはカラム指向のストレージフォーマットであり、分析ワークロードにおいて高い圧縮率とI/O効率を実現する。Sparkは述語プッシュダウン、カラムプルーニング、ベクトル化読み取り等の最適化を通じて、Parquetデータへの高性能なアクセスを提供する。

**業務上の目的・背景**：大規模データ分析において、行指向フォーマット（CSV等）では不要なカラムも読み込む必要があり、I/Oが非効率であった。Parquetカラムナフォーマットは、必要なカラムのみを読み込むカラムプルーニング、条件に合わないデータをスキップする述語プッシュダウン、高い圧縮率によるストレージ節約を実現する。Sparkはこれらの最適化を自動的に適用し、大規模データの分析処理を高速化する。

**機能の利用シーン**：データレイク上のデータの永続化・読み込み、ETLパイプラインの中間データ格納、データウェアハウスのテーブルフォーマット、機械学習の訓練データ格納。

**主要な処理内容**：
1. Parquetファイルの読み込み（スキーマ推論、述語プッシュダウン、カラムプルーニング）
2. Parquetファイルの書き出し（圧縮コーデック選択、パーティション書き込み）
3. スキーママージ（複数Parquetファイル間のスキーマ統合）
4. ベクトル化読み取り（ColumnarBatch対応）
5. フィルタプッシュダウン（Parquetの統計情報を利用した行グループスキップ）
6. 型変換と日時型のリベース処理

**関連システム・外部連携**：Apache Parquetライブラリ（parquet-hadoop）、Hadoop FileSystem API、Catalystオプティマイザ（プッシュダウン最適化）

**権限による制御**：ファイルシステムのアクセス権限に依存。HDFS ACLやクラウドストレージのIAMポリシーにより制御される。

## 関連画面

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

## 機能種別

データ連携 / CRUD操作 / ファイルI/O

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| path | String | Yes | Parquetファイルパス | 有効なファイルシステムパス |
| mergeSchema | Boolean | No | スキーママージの有効/無効（デフォルト: false） | true/false |
| compression | String | No | 圧縮コーデック（snappy, gzip, lzo, zstd, none） | サポート対象コーデック |
| spark.sql.parquet.filterPushdown | Boolean | No | フィルタプッシュダウン（デフォルト: true） | true/false |
| spark.sql.parquet.enableVectorizedReader | Boolean | No | ベクトル化リーダー（デフォルト: true） | true/false |

### 入力データソース

- Hadoop互換ファイルシステム上のParquetファイル（HDFS, S3, GCS, Azure Blob等）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| DataFrame | Dataset[Row] | 読み込まれたParquetデータ |
| InternalRow | InternalRow | 行単位の内部データ表現 |
| ColumnarBatch | ColumnarBatch | ベクトル化読み取り時のバッチデータ |

### 出力先

- DataFrame/Dataset（読み取り結果）
- Hadoop互換ファイルシステム上のParquetファイル（書き出し結果）

## 処理フロー

### 処理シーケンス

```
1. ParquetFileFormat/ParquetDataSourceV2によるデータソース登録
   └─ shortName "parquet" として登録
2. 読み込み時: スキーマ推論またはユーザー指定スキーマの適用
   └─ ParquetUtils.inferSchema()によるParquetメタデータからのスキーマ取得
3. フィルタプッシュダウンの適用
   └─ ParquetFilters.createFilter()でSpark述語をParquetフィルタに変換
4. カラムプルーニングの適用
   └─ 必要カラムのみをリクエストスキーマに設定
5. ファイル読み取り実行
   └─ ベクトル化リーダーまたは行リーダーによるデータ読み取り
6. 書き出し時: ParquetOutputWriter による書き込み
   └─ 圧縮コーデック適用、パーティション書き込み
```

### フローチャート

```mermaid
flowchart TD
    A[spark.read.parquet] --> B[ParquetFileFormat]
    B --> C{スキーマ指定?}
    C -->|Yes| D[ユーザースキーマ適用]
    C -->|No| E[スキーマ推論]
    D --> F[フィルタプッシュダウン]
    E --> F
    F --> G[カラムプルーニング]
    G --> H{ベクトル化対応?}
    H -->|Yes| I[VectorizedParquetRecordReader]
    H -->|No| J[ParquetRecordReader]
    I --> K[ColumnarBatch]
    J --> L[InternalRow]
    K --> M[DataFrame]
    L --> M
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-24-01 | デフォルトフォーマット | spark.sql.sources.defaultのデフォルト値が"parquet" | 明示的フォーマット指定なし時 |
| BR-24-02 | 圧縮コーデック | デフォルト圧縮はsnappy | 書き出し時 |
| BR-24-03 | スキーママージ | mergeSchema=trueで複数ファイルのスキーマを統合 | 読み込み時 |
| BR-24-04 | ベクトル化リーダー制約 | ネストした型やUDTを含む場合はベクトル化リーダーが使用されない | 読み込み時 |

### 計算ロジック

フィルタプッシュダウンでは、Parquetファイルの行グループ（RowGroup）ごとの統計情報（min/max値）を利用して、条件に合致しない行グループ全体をスキップする。これにより、実際に読み込むデータ量を大幅に削減できる。

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

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

本機能はファイルシステム上のParquetファイルを対象とし、直接的なデータベース操作は行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| CANNOT_READ_FILE_FOOTER | 読み取りエラー | Parquetファイルのフッターが破損 | ファイルの再生成 |
| SCHEMA_MISMATCH | スキーマエラー | スキーマ進化で互換性のない変更 | スキーママージの確認 |
| UNSUPPORTED_DATA_TYPE | 型エラー | Sparkでサポートされない型を含むParquetファイル | 型マッピングの確認 |

### リトライ仕様

ファイル読み取りの失敗はタスクレベルでリトライされる。

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

書き出し時はFileOutputCommitterによるコミットプロトコルが適用される。タスクの成功/失敗に応じて部分的な出力が管理される。

## パフォーマンス要件

- カラムプルーニングによる不要データの読み飛ばし
- 述語プッシュダウンによる行グループスキップ
- ベクトル化読み取りによるCPU効率向上
- Snappy/ZSTD圧縮によるI/O削減
- パーティション刈り込み（Partition Pruning）

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

- ファイルシステムのアクセス権限に依存
- Parquetファイル自体に暗号化機能（Parquet Modular Encryption）が利用可能

## 備考

- Sparkのデフォルトデータソースフォーマットはparquet
- Parquet V1/V2の書き込みフォーマットが選択可能

---

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

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

### 推奨読解順序

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

Parquetサポートの設定とオプション構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ParquetOptions.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetOptions.scala` | 圧縮コーデック等のParquet固有オプション |

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

ParquetFileFormatがV1データソースの主要実装である。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ParquetFileFormat.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala` | FileFormatインターフェースの実装 |

**主要処理フロー**:
1. **58-63行目**: ParquetFileFormatクラス定義、FileFormat/DataSourceRegisterの実装
2. **65行目**: shortName() = "parquet"
3. **73-80行目**: prepareWrite() - 書き込み準備、ParquetOptionsとParquetUtils.prepareWrite使用

#### Step 3: 読み取り処理を理解する

ParquetUtilsとフィルタ変換を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ParquetUtils.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala` | スキーマ推論、読み取りユーティリティ |
| 3-2 | ParquetFilters.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala` | Spark述語からParquetフィルタへの変換 |

#### Step 4: 書き込み処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | ParquetOutputWriter.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetOutputWriter.scala` | Parquetファイルの書き込み |

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

```
DataFrameReader.parquet() / spark.read.parquet()
    │
    ├─ ParquetFileFormat
    │      ├─ inferSchema() → ParquetUtils.inferSchema()
    │      ├─ prepareWrite() → ParquetUtils.prepareWrite()
    │      └─ buildReaderWithPartitionValues()
    │             ├─ ParquetFilters.createFilter() (述語プッシュダウン)
    │             ├─ VectorizedParquetRecordReader (ベクトル化)
    │             └─ ParquetRecordReader (非ベクトル化)
    │
    └─ ParquetOutputWriter
           └─ ParquetWriter (Apache Parquetライブラリ)
```

### データフロー図

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

Parquetファイル ─────▶ ParquetFileFormat ──────▶ DataFrame
(HDFS/S3/GCS)           │
                        ├─ スキーマ推論
                        ├─ フィルタプッシュダウン
                        ├─ カラムプルーニング
                        └─ ベクトル化読み取り

DataFrame ──────────▶ ParquetOutputWriter ──────▶ Parquetファイル
                        │
                        ├─ 圧縮コーデック適用
                        └─ パーティション書き込み
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ParquetFileFormat.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala` | ソース | V1 Parquetデータソース実装 |
| ParquetUtils.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala` | ソース | ユーティリティ関数 |
| ParquetFilters.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala` | ソース | フィルタプッシュダウン変換 |
| ParquetOptions.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetOptions.scala` | ソース | Parquetオプション定義 |
| ParquetOutputWriter.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetOutputWriter.scala` | ソース | 書き込み処理 |
| ParquetSchemaConverter.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala` | ソース | スキーマ変換 |
