# 機能設計書 38-ファイルコネクタ

## 概要

本ドキュメントは、Apache Flinkのファイルコネクタ（FileSink）機能について記載する。この機能は、ストリームデータをファイルシステムに書き込むための統一的なシンクコネクタであり、BATCHとSTREAMINGの両モードでExactly-onceセマンティクスを提供する。

### 本機能の処理概要

FileSinkは、入力要素をFileSystemファイルにバケット単位で書き込む統一シンクである。ベースディレクトリ配下にバケットディレクトリを作成し、各サブタスクごとにパートファイルを生成する。行形式（Row Format）とバルク形式（Bulk Format：Parquet/ORC等）の両方をサポートする。

**業務上の目的・背景**：ストリーム処理の結果をファイルシステム（HDFS、S3、GCS等）に永続化する必要がある。FileSinkは、チェックポイントと連携してExactly-once保証を実現し、障害時のデータ整合性を確保する。また、ファイルコンパクション機能により、小さなファイルを統合してクエリパフォーマンスを向上させる。

**機能の利用シーン**：
- ストリームデータのデータレイクへの書き込み
- ETLパイプラインの出力
- ログデータのアーカイブ
- 時間ベースのパーティショニング

**主要な処理内容**：
1. BucketAssignerによるバケット割り当て
2. RollingPolicyによるファイルローリング
3. In-progress/Pending/Finishedファイル状態管理
4. FileCommitterによるファイルコミット
5. FileCompactorによるファイルコンパクション

**関連システム・外部連携**：
- FileSystem（HDFS、S3、GCS、ローカルファイルシステム等）
- CheckpointStorage
- OutputFileConfig

**権限による制御**：ファイルシステムの権限に依存。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は画面とは直接関連しない（バックエンド処理） |

## 機能種別

データ出力 / ファイル書き込み / コネクタ

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| basePath | Path | Yes | ベースディレクトリパス | 有効なパス |
| encoder | Encoder<IN> | Row時Yes | 行エンコーダー | 非null |
| bulkWriterFactory | BulkWriter.Factory<IN> | Bulk時Yes | バルクライターファクトリ | 非null |
| bucketAssigner | BucketAssigner | No | バケット割り当て | デフォルトDateTimeBucketAssigner |
| rollingPolicy | RollingPolicy | No | ローリングポリシー | デフォルトDefaultRollingPolicy |
| bucketCheckInterval | long | No | バケットチェック間隔 | デフォルト60秒 |
| outputFileConfig | OutputFileConfig | No | 出力ファイル設定 | デフォルト設定 |

### 入力データソース

- DataStream<IN>

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Part Files | File | パートファイル（出力データ） |
| In-progress Files | File | 書き込み中ファイル |
| Pending Files | File | コミット待ちファイル |

### 出力先

- FileSystem（basePath配下のバケットディレクトリ）

## 処理フロー

### 処理シーケンス

```
1. FileSink初期化
   └─ BucketsBuilder設定

2. FileWriter作成
   └─ BucketAssigner、RollingPolicy、BucketWriter設定

3. データ書き込み
   └─ バケット割り当て
   └─ パートファイル書き込み

4. ファイルローリング
   └─ RollingPolicyに基づくファイル切り替え
   └─ In-progress → Pending

5. チェックポイント/ジョブ終了時
   └─ FileCommitterによるコミット
   └─ Pending → Finished

6. コンパクション（有効時）
   └─ CompactorCoordinatorでリクエスト生成
   └─ CompactorOperatorで実行
```

### フローチャート

```mermaid
flowchart TD
    A[データ受信] --> B[BucketAssigner]
    B --> C[バケット決定]
    C --> D[FileWriter]
    D --> E[In-progress File書き込み]
    E --> F{Rolling条件?}
    F -->|Yes| G[ファイルローリング]
    G --> H[Pending状態]
    F -->|No| E
    H --> I{チェックポイント?}
    I -->|Yes| J[FileCommitter]
    J --> K[Finished状態]
    I -->|No| H
    K --> L{Compact有効?}
    L -->|Yes| M[CompactorCoordinator]
    M --> N[CompactorOperator]
    N --> O[コンパクト済みファイル]
    L -->|No| P[完了]
    O --> P
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-38-01 | Exactly-once | BATCH/STREAMINGでExactly-once保証 | 常時 |
| BR-38-02 | ファイル状態遷移 | in-progress → pending → finished | ファイルライフサイクル |
| BR-38-03 | リカバリ動作 | リストア時、pendingはfinishedに、in-progressはロールバック | 障害復旧時 |
| BR-38-04 | デフォルトバケット | DateTimeBucketAssignerで1時間毎にバケット作成 | バケット未指定時 |

### 計算ロジック

ファイル名生成: `{prefix}-{uid}-{counter}{suffix}`
- prefix: OutputFileConfigのpartPrefix
- uid: サブタスクのUID
- counter: ローリングカウンター
- suffix: OutputFileConfigのpartSuffix

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | RDBMSは使用しない |

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

該当なし（FileSystemに出力）

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IO | IOException | ファイル書き込み失敗 | ファイルシステム設定確認 |
| COMMIT | Exception | コミット失敗 | リトライまたはチェックポイントからリストア |
| COMPACT | Exception | コンパクション失敗 | コンパクション設定確認 |

### リトライ仕様

チェックポイントメカニズムによる自動リカバリ

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

- RecoverableWriterによるトランザクショナル書き込み
- チェックポイント成功時にファイルコミット

## パフォーマンス要件

- bucketCheckInterval: デフォルト60秒
- RollingPolicy: ファイルサイズ、時間、チェックポイントベース

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

ファイルシステムへのアクセスには適切な認証・認可設定が必要

## 備考

- @Experimental
- StreamingFileSinkとの後方互換性あり（bucket-statesステート名）

---

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

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

### 推奨読解順序

#### Step 1: FileSinkクラスを理解する

メインクラスの構造とビルダーパターンを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | FileSink.java | `flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/` | メインクラス、Sink API実装 |

**読解のコツ**:
- **79-132行目**: クラスコメントで設計意図を把握
- **134-140行目**: 実装インターフェース（Sink、SupportsWriterState、SupportsCommitter等）
- **148-160行目**: createWriter()、restoreWriter()
- **203-212行目**: forRowFormat()、forBulkFormat()ファクトリメソッド

#### Step 2: RowFormatBuilderとBulkFormatBuilderを理解する

行形式とバルク形式のビルダーを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | FileSink.java | 同上 | 332-504行目 RowFormatBuilder |
| 2-2 | FileSink.java | 同上 | 508-697行目 BulkFormatBuilder |

**読解のコツ**:
- **357-367行目**: RowFormatBuilderコンストラクタ、デフォルト設定
- **386-404行目**: withBucketCheckInterval、withBucketAssigner、withRollingPolicy等の設定メソッド
- **406-415行目**: enableCompact()、disableCompact()
- **489-493行目**: createBucketWriter() RowWiseBucketWriter生成

#### Step 3: FileWriterを理解する

ライター実装を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | FileWriter.java | `flink-connector-files/.../sink/writer/` | ライター実装 |
| 3-2 | FileWriterBucket.java | 同パス | バケット管理 |

#### Step 4: コンパクション機能を理解する

ファイルコンパクションの実装を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | CompactCoordinator.java | `flink-connector-files/.../compactor/operator/` | コンパクションコーディネータ |
| 4-2 | CompactorOperator.java | 同パス | コンパクション実行 |

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

```
FileSink
    │
    ├─ implements Sink, SupportsWriterState, SupportsCommitter
    │
    ├─ forRowFormat(basePath, encoder)
    │      └─ DefaultRowFormatBuilder
    │             ├─ withBucketAssigner()
    │             ├─ withRollingPolicy()
    │             ├─ withOutputFileConfig()
    │             ├─ enableCompact()
    │             └─ build() → FileSink
    │
    ├─ forBulkFormat(basePath, bulkWriterFactory)
    │      └─ DefaultBulkFormatBuilder
    │             ├─ withBucketAssigner()
    │             ├─ withRollingPolicy()
    │             └─ build() → FileSink
    │
    ├─ createWriter(context)
    │      └─ FileWriter
    │             ├─ BucketAssigner
    │             ├─ RollingPolicy
    │             └─ BucketWriter（Row/Bulk）
    │
    ├─ createCommitter(context)
    │      └─ FileCommitter
    │
    └─ addPreCommitTopology()（コンパクション有効時）
           ├─ CompactCoordinatorFactory
           │      └─ CompactCoordinator
           └─ CompactorOperatorFactory
                  └─ CompactorOperator
```

### データフロー図

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

DataStream<IN> ─────────▶ FileSink ──────────────────▶ FileSystem
                               │
                               ├─ FileWriter
                               │      ├─ BucketAssigner
                               │      │      └─ バケット決定
                               │      ├─ RollingPolicy
                               │      │      └─ ローリング判定
                               │      └─ BucketWriter
                               │             └─ パートファイル書き込み
                               │
                               ├─ FileCommitter
                               │      └─ Pending → Finished
                               │
                               └─ CompactorOperator（オプション）
                                      └─ 小ファイル統合

ファイル状態遷移:
In-progress ──▶ Pending ──▶ Finished
   (書込中)      (待機)       (完了)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| FileSink.java | `flink-connector-files/.../sink/` | メインクラス | シンク定義 |
| FileWriter.java | `flink-connector-files/.../sink/writer/` | ライター | データ書き込み |
| FileWriterBucket.java | `flink-connector-files/.../sink/writer/` | バケット | バケット管理 |
| FileWriterBucketState.java | `flink-connector-files/.../sink/writer/` | 状態 | ライター状態 |
| FileCommitter.java | `flink-connector-files/.../sink/committer/` | コミッター | ファイルコミット |
| FileSinkCommittable.java | `flink-connector-files/.../sink/` | コミッタブル | コミット情報 |
| CompactCoordinator.java | `flink-connector-files/.../compactor/operator/` | コーディネータ | コンパクション調整 |
| CompactorOperator.java | `flink-connector-files/.../compactor/operator/` | オペレータ | コンパクション実行 |
| FileCompactStrategy.java | `flink-connector-files/.../compactor/` | ストラテジー | コンパクション戦略 |
| BucketAssigner.java | `flink-streaming-java/.../filesystem/` | インターフェース | バケット割り当て |
| RollingPolicy.java | `flink-streaming-java/.../filesystem/` | インターフェース | ローリングポリシー |
| OutputFileConfig.java | `flink-streaming-java/.../filesystem/` | 設定 | 出力ファイル設定 |
