# 機能設計書 51-S3（Hadoop）

## 概要

本ドキュメントは、Apache Flink の Amazon S3 ファイルシステム（Hadoopベース）の機能設計書である。flink-s3-fs-hadoop モジュールが提供する S3 ストレージとの連携機能について、処理内容、設定方法、およびコードの構造を詳細に記載する。

### 本機能の処理概要

**業務上の目的・背景**：ビッグデータ処理において、大規模なデータを安価かつ耐久性の高いストレージに保存することが求められる。Amazon S3 は業界標準のオブジェクトストレージであり、Flink ジョブの入出力データ、チェックポイント、セーブポイントの保存先として広く利用される。本機能により、Flink は Hadoop の S3A ファイルシステム実装を活用し、S3 へのシームレスなアクセスを提供する。

**機能の利用シーン**：
- ストリーム処理ジョブのチェックポイントを S3 に保存する場合
- バッチ処理ジョブの入力データを S3 から読み込む場合
- 処理結果を S3 バケットに出力する場合
- 高可用性（HA）設定でメタデータを S3 に保存する場合

**主要な処理内容**：
1. S3FileSystemFactory による FileSystem インスタンスの生成
2. Hadoop Configuration の読み込みとFlinkパラメータのマッピング
3. S3AFileSystem の初期化とラッピング
4. マルチパートアップロードによる大容量ファイルの効率的な書き込み
5. RecoverableWriter によるチェックポイント対応の書き込み

**関連システム・外部連携**：
- Amazon S3 / S3互換ストレージ（MinIO等）
- AWS SDK for Java
- Hadoop S3A ファイルシステム

**権限による制御**：AWS IAM によるアクセス制御。アクセスキー/シークレットキーまたは IAM ロールによる認証をサポート。

## 関連画面

本機能は画面を持たないバックエンド機能である。

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

## 機能種別

データ連携 / ファイルシステム抽象化

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| fs.s3a.access-key | String | Yes（環境変数/IAMロール未使用時） | AWSアクセスキー | 非空文字列 |
| fs.s3a.secret-key | String | Yes（環境変数/IAMロール未使用時） | AWSシークレットキー | 非空文字列 |
| fs.s3a.endpoint | String | No | S3互換エンドポイント | 有効なURL |
| fs.s3a.path-style-access | Boolean | No | パススタイルアクセスの有効化 | true/false |
| s3.entropy.key | String | No | エントロピー注入キー | 文字列 |
| s3.entropy.length | Integer | No | エントロピー文字数 | 正の整数 |

### 入力データソース

- Flink Configuration（flink-conf.yaml）
- 環境変数（AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY）
- Hadoop Configuration（core-site.xml）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| FileSystem | org.apache.flink.core.fs.FileSystem | Flinkファイルシステムインターフェース実装 |

### 出力先

Flink ランタイムへの FileSystem インスタンス提供

## 処理フロー

### 処理シーケンス

```
1. FileSystemFactory.create(URI) が呼び出される
   └─ URIスキーム（s3:// または s3a://）を検証
2. HadoopConfigLoader で設定を読み込み
   └─ Flink設定からHadoop設定へのマッピング
3. S3AFileSystem インスタンスを生成
   └─ new S3AFileSystem()
4. S3AFileSystem.initialize(URI, Configuration) で初期化
   └─ AWS認証情報の設定
5. HadoopS3AccessHelper を生成（RecoverableWriter用）
   └─ マルチパートアップロード機能の提供
6. FlinkS3FileSystem でラップして返却
   └─ Flink FileSystem インターフェースへの適合
```

### フローチャート

```mermaid
flowchart TD
    A[開始: create URI] --> B{URIスキーム確認}
    B -->|s3:// or s3a://| C[HadoopConfigLoader実行]
    B -->|その他| Z[エラー: 非対応スキーム]
    C --> D[S3AFileSystem生成]
    D --> E[Configuration適用]
    E --> F[S3AFileSystem.initialize]
    F --> G{初期化成功?}
    G -->|Yes| H[HadoopS3AccessHelper生成]
    G -->|No| Z2[エラー: 初期化失敗]
    H --> I[FlinkS3FileSystem生成]
    I --> J[終了: FileSystem返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-51-01 | スキームマッピング | s3:// と s3a:// の両方を S3AFileSystem にマッピング | 常時 |
| BR-51-02 | 設定プレフィックス | s3., s3a., fs.s3a. プレフィックスの設定を読み込み | 設定読み込み時 |
| BR-51-03 | 認証優先順位 | 明示的設定 > 環境変数 > IAMロール | 認証情報解決時 |

### 計算ロジック

特になし

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

本機能はデータベース操作を行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | 接続エラー | S3エンドポイントに接続できない | ネットワーク設定を確認 |
| AccessDeniedException | 認証エラー | 認証情報が無効または権限不足 | IAM設定を確認 |
| IllegalArgumentException | 設定エラー | 必須設定が欠落 | flink-conf.yamlを確認 |

### リトライ仕様

S3A ファイルシステムの標準リトライポリシーに従う（エクスポネンシャルバックオフ）

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

オブジェクトストレージのため、トランザクションは非対応。マルチパートアップロードのアボート処理により部分書き込みを防止。

## パフォーマンス要件

- マルチパートアップロードによる並列書き込み
- 推奨パートサイズ: 64MB以上
- 接続プールによるコネクション再利用

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

- 認証情報は環境変数またはIAMロールでの提供を推奨
- 設定ファイルへの認証情報直接記載は非推奨
- HTTPS接続の使用を推奨（fs.s3a.connection.ssl.enabled=true）

## 備考

- S3互換ストレージ（MinIO等）との接続には fs.s3a.endpoint の設定が必要
- パススタイルアクセスは古いS3互換ストレージで必要な場合がある

---

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

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

### 推奨読解順序

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

S3ファイルシステムの設定とアクセスヘルパーの構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | HadoopConfigLoader.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/util/HadoopConfigLoader.java` | Flink設定からHadoop設定への変換ロジック |

**読解のコツ**: FLINK_CONFIG_PREFIXES と MIRRORED_CONFIG_KEYS の配列定義を確認し、どの設定キーがどのようにマッピングされるかを把握する。

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

ファイルシステムファクトリの実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | S3FileSystemFactory.java | `flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3FileSystemFactory.java` | ファクトリのメイン実装 |
| 2-2 | S3AFileSystemFactory.java | `flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3AFileSystemFactory.java` | s3a:// スキーム用のサブクラス |

**主要処理フロー**:
1. **37-51行目**: コンストラクタでHadoopConfigLoaderを初期化
2. **54-56行目**: getScheme()で"s3"を返却
3. **70-72行目**: createHadoopFileSystem()でS3AFileSystemを生成
4. **75-91行目**: getInitURI()でURIの正規化
5. **94-98行目**: getS3AccessHelper()でアクセスヘルパーを生成

#### Step 3: アクセスヘルパーを理解する

マルチパートアップロード機能の実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | HadoopS3AccessHelper.java | `flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/HadoopS3AccessHelper.java` | S3操作の実装 |

**主要処理フロー**:
- **57-67行目**: コンストラクタでWriteOperationHelperを初期化
- **70-72行目**: startMultiPartUpload()でマルチパートアップロード開始
- **75-88行目**: uploadPart()でパートのアップロード
- **91-94行目**: putObject()で単一オブジェクトのアップロード
- **97-106行目**: commitMultiPartUpload()でアップロード完了

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

```
S3FileSystemFactory.create(URI)
    │
    ├─ createHadoopConfigLoader()
    │      └─ HadoopConfigLoader(prefixes, mirroredKeys, ...)
    │
    ├─ createHadoopFileSystem()
    │      └─ new S3AFileSystem()
    │
    ├─ getInitURI(fsUri, hadoopConfig)
    │      └─ FileSystem.getDefaultUri() [if needed]
    │
    ├─ S3AFileSystem.initialize(URI, Configuration)
    │
    └─ getS3AccessHelper(S3AFileSystem)
           └─ new HadoopS3AccessHelper(s3a, conf)
                  └─ new InternalWriteOperationHelper(...)
```

### データフロー図

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

URI (s3://bucket/path) ──▶ S3FileSystemFactory ──────▶ FlinkS3FileSystem
                               │
Flink Configuration ────────▶ HadoopConfigLoader ───▶ Hadoop Configuration
                               │
AWS Credentials ────────────▶ S3AFileSystem ─────────▶ S3 API Calls
                               │
                          HadoopS3AccessHelper ──────▶ Multipart Upload
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| S3FileSystemFactory.java | `flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3FileSystemFactory.java` | ソース | s3://スキーム用ファクトリ |
| S3AFileSystemFactory.java | `flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3AFileSystemFactory.java` | ソース | s3a://スキーム用ファクトリ |
| HadoopS3AccessHelper.java | `flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/HadoopS3AccessHelper.java` | ソース | S3アクセスヘルパー |
| HadoopConfigLoader.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/util/HadoopConfigLoader.java` | ソース | 設定ローダー |
| AbstractS3FileSystemFactory.java | `flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/AbstractS3FileSystemFactory.java` | ソース | 基底ファクトリクラス |
| org.apache.flink.core.fs.FileSystemFactory | `flink-filesystems/flink-s3-fs-hadoop/src/main/resources/META-INF/services/` | 設定 | SPIサービス登録 |
