# 機能設計書 48-圧縮

## 概要

本ドキュメントは、Apache Flink の flink-compress モジュールが提供するデータ圧縮書き込み機能の設計を記述する。

### 本機能の処理概要

圧縮フォーマット機能は、Hadoop CompressionCodec を使用してデータを圧縮しながらファイルに書き込むための BulkWriter を提供する。GZIP、Snappy、LZ4 などの各種圧縮コーデックをサポートする。

**業務上の目的・背景**：大規模データ処理において、ストレージコストと I/O 帯域幅の削減は重要な要素である。圧縮により、ストレージ使用量を削減しつつ、データ転送時間を短縮できる。Hadoop エコシステムとの互換性を維持しながら、柔軟な圧縮オプションを提供する。

**機能の利用シーン**：
- 大規模ログファイルの圧縮保存
- ETL パイプラインでのストレージ最適化
- アーカイブデータの圧縮
- ネットワーク転送前のデータ圧縮
- コスト効率の良いデータレイク構築

**主要な処理内容**：
1. CompressWriterFactory: 圧縮ライターのファクトリ
2. HadoopCompressionBulkWriter: Hadoop コーデックによる圧縮書き込み
3. NoCompressionBulkWriter: 非圧縮書き込み
4. Extractor: データからバイト配列への変換

**関連システム・外部連携**：
- ファイルシステムコネクタ
- Hadoop CompressionCodec（GZIP、Snappy、LZ4 等）
- HDFS / クラウドストレージ

**権限による制御**：本機能はファイル書き込み処理のため、権限制御はファイルシステムに委譲される。

## 関連画面

本機能はバックエンドのデータ書き込み機能であり、直接関連する画面はない。

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

## 機能種別

- ファイル書き込み
- データ圧縮

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| codecName | String | No | Hadoop CompressionCodec 名 | 有効なコーデック名 |
| hadoopConfig | Configuration | No | Hadoop 設定 | - |
| extractor | Extractor<IN> | Yes | データ抽出関数 | null 不可 |

### 入力データソース

- 書き込み: 任意の型のデータ（Extractor で byte[] に変換）

### サポートされる圧縮コーデック

| コーデック名 | 拡張子 | 説明 |
|------------|--------|------|
| gzip | .gz | 標準的な圧縮、高圧縮率 |
| snappy | .snappy | 高速圧縮、中程度の圧縮率 |
| lz4 | .lz4 | 非常に高速な圧縮 |
| bzip2 | .bz2 | 最高圧縮率、低速 |
| deflate | .deflate | zlib ベースの圧縮 |

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| compressedFile | File | 圧縮されたファイル |

### 出力先

- ファイルシステム（HDFS、S3、ローカル等）

## 処理フロー

### フローチャート

```mermaid
flowchart TD
    A[Input Data] --> B[Extractor]
    B --> C{Codec specified?}
    C -->|Yes| D[HadoopCompressionBulkWriter]
    C -->|No| E[NoCompressionBulkWriter]
    D --> F[CompressionOutputStream]
    F --> G[Compressed File]
    E --> H[FSDataOutputStream]
    H --> I[Uncompressed File]
```

### データフロー図

```mermaid
flowchart LR
    subgraph Input
        Data[Input Element]
    end

    subgraph Processing
        Ext[Extractor.extract]
        Ext --> Bytes[byte[]]
    end

    subgraph Compression
        Codec[CompressionCodec]
        Codec --> COS[CompressionOutputStream]
    end

    subgraph Output
        File[Compressed File]
    end

    Data --> Ext
    Bytes --> COS
    COS --> File
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-48-01 | コーデック選択 | codecName 未指定時は非圧縮で書き込み | 書き込み時 |
| BR-48-02 | 拡張子付与 | コーデックのデフォルト拡張子をファイルに付与 | 圧縮時 |
| BR-48-03 | コーデック遅延初期化 | CompressionCodec は初回使用時に初期化 | 書き込み開始時 |
| BR-48-04 | Hadoop 設定継承 | Hadoop Configuration をシリアライズして保持 | 初期化時 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | コーデック未発見 | 無効なコーデック名 | コーデック名確認 |
| IOException | 書き込みエラー | ファイル書き込み失敗 | ファイルパス・権限確認 |
| NullPointerException | Extractor null | Extractor 未指定 | Extractor 指定 |

## 備考

- Hadoop CompressionCodec を直接使用
- 書き込み専用（読み取りはサポートしない）
- StreamingFileSink との組み合わせで使用

---

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

### 推奨読解順序

#### Step 1: ビルダークラス

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | CompressWriters.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/CompressWriters.java` | エントリーポイント |

**主要処理**:
- **26-28行目**: forExtractor() で CompressWriterFactory 生成

#### Step 2: ファクトリクラス

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | CompressWriterFactory.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/CompressWriterFactory.java` | ファクトリ実装 |

**主要処理フロー**:
- **63-66行目**: コンストラクタで Extractor 設定
- **75-77行目**: withHadoopCompression() でコーデック設定（デフォルト Configuration）
- **88-98行目**: withHadoopCompression() でコーデック設定（カスタム Configuration）
- **101-109行目**: create() で BulkWriter 生成
- **102-104行目**: コーデック未指定時は NoCompressionBulkWriter
- **106-108行目**: コーデック指定時は HadoopCompressionBulkWriter
- **115-125行目**: initializeCompressionCodec() で遅延初期化
- **127-137行目**: getHadoopCodecExtension() でコーデック拡張子取得

#### Step 3: BulkWriter 実装

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | HadoopCompressionBulkWriter.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/writers/HadoopCompressionBulkWriter.java` | 圧縮書き込み |
| 3-2 | NoCompressionBulkWriter.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/writers/NoCompressionBulkWriter.java` | 非圧縮書き込み |

**圧縮書き込み処理**:
- **41-44行目**: コンストラクタで CompressionOutputStream と Extractor 設定
- **47-49行目**: addElement() で要素を圧縮書き込み
- **52-54行目**: flush() でバッファフラッシュ
- **57-59行目**: finish() で圧縮完了

#### Step 4: Extractor インターフェース

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | Extractor.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/extractor/Extractor.java` | データ抽出インターフェース |

**インターフェース定義**:
- **27-29行目**: extract() メソッドで要素から byte[] への変換

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

```
CompressWriters
└── forExtractor()
    └── CompressWriterFactory
        ├── withHadoopCompression()
        │   ├── getHadoopCodecExtension()
        │   └── CompressionCodecFactory.getCodecByName()
        └── create()
            ├── NoCompressionBulkWriter (codecName == null)
            │   └── Extractor.extract()
            └── HadoopCompressionBulkWriter (codecName != null)
                ├── initializeCompressionCodec()
                ├── CompressionCodec.createOutputStream()
                └── Extractor.extract()
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| CompressWriters.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/CompressWriters.java` | ソース | ビルダー |
| CompressWriterFactory.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/CompressWriterFactory.java` | ソース | ファクトリ |
| HadoopCompressionBulkWriter.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/writers/HadoopCompressionBulkWriter.java` | ソース | 圧縮ライター |
| NoCompressionBulkWriter.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/writers/NoCompressionBulkWriter.java` | ソース | 非圧縮ライター |
| Extractor.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/extractor/Extractor.java` | ソース | 抽出インターフェース |
| DefaultExtractor.java | `flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/extractor/DefaultExtractor.java` | ソース | デフォルト抽出実装 |
