# 機能設計書 56-HDFS

## 概要

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

### 本機能の処理概要

**業務上の目的・背景**：Hadoop エコシステムは多くの企業でビッグデータ基盤として採用されている。HDFS は分散ファイルシステムとして高いスケーラビリティと耐障害性を提供する。Flink は HDFS と統合することで、既存の Hadoop 基盤上でストリーム処理・バッチ処理を実行でき、データレイクとのシームレスな連携が可能となる。

**機能の利用シーン**：
- ストリーム処理ジョブのチェックポイントを HDFS に保存する場合
- バッチ処理ジョブの入力データを HDFS から読み込む場合
- 処理結果を HDFS に出力する場合
- 既存 Hadoop クラスタとの統合

**主要な処理内容**：
1. HadoopFsFactory による FileSystem インスタンスの生成
2. Hadoop Configuration の読み込みと適用
3. 任意のスキーム（hdfs://, viewfs:// 等）のサポート
4. HadoopFileSystem による org.apache.hadoop.fs.FileSystem のラッピング
5. HadoopRecoverableWriter によるチェックポイント対応書き込み

**関連システム・外部連携**：
- Hadoop Distributed File System（HDFS）
- Hadoop ViewFS（フェデレーテッドHDFS）
- その他 Hadoop 互換ファイルシステム

**権限による制御**：Hadoop セキュリティ（Kerberos）、HDFS ACL によるアクセス制御をサポート。

## 関連画面

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

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

## 機能種別

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

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| fs.defaultFS | String | No | デフォルトファイルシステムURI | 有効なURI |
| fs.hdfs.hadoopconf | String | No | Hadoop設定ディレクトリ | 有効なディレクトリパス |
| fs.hdfs.no-local-write | Boolean | No | ローカル書き込みの無効化 | true/false |

### 入力データソース

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

## 出力仕様

### 出力データ

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

### 出力先

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

## 処理フロー

### 処理シーケンス

```
1. FileSystemFactory.create(URI) が呼び出される
   └─ URIスキームを取得（hdfs://, viewfs:// 等）
2. Hadoop Configuration を読み込み
   └─ HadoopUtils.getHadoopConfiguration()
3. Hadoop FileSystem クラスを取得
   └─ FileSystem.getFileSystemClass(scheme, hadoopConfig)
4. Hadoop FileSystem インスタンスを生成・初期化
5. HadoopFileSystem でラップ
6. LimitedConnectionsFileSystem で接続制限を適用（オプション）
7. FileSystem を返却
```

### フローチャート

```mermaid
flowchart TD
    A[開始: create URI] --> B[スキーム取得]
    B --> C[Hadoop Configuration読み込み]
    C --> D[FileSystem.getFileSystemClass]
    D --> E{クラス取得成功?}
    E -->|Yes| F[FileSystemインスタンス生成]
    E -->|No| Z[UnsupportedFileSystemSchemeException]
    F --> G[URI authority確認]
    G --> H{authority有り?}
    H -->|Yes| I[fsUri使用]
    H -->|No| J[fs.defaultFS使用]
    I --> K[initialize実行]
    J --> K
    K --> L[HadoopFileSystem生成]
    L --> M{接続制限設定?}
    M -->|Yes| N[LimitedConnectionsFileSystem適用]
    M -->|No| O[終了: FileSystem返却]
    N --> O
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-56-01 | ワイルドカードスキーム | HadoopFsFactory は "*" スキームで登録（任意のスキーム対応） | 常時 |
| BR-56-02 | デフォルトURI | authority未指定時は fs.defaultFS を使用 | URI解決時 |
| BR-56-03 | ファイルシステム種別判定 | スキームに基づいて OBJECT_STORE または FILE_SYSTEM を判定 | getKindForScheme() |
| BR-56-04 | 接続制限 | 設定に基づいて接続数を制限 | LimitedConnectionsFileSystem 適用時 |

### 計算ロジック

特になし

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| UnsupportedFileSystemSchemeException | スキーム非対応 | 指定スキームのファイルシステムがクラスパスにない | 必要な依存関係を追加 |
| UnknownHostException | ホスト解決エラー | NameNode に接続できない | ネットワーク・DNS設定を確認 |
| IOException | 接続エラー | HDFS への接続に失敗 | 設定・ネットワークを確認 |

### リトライ仕様

Hadoop ファイルシステムの標準リトライポリシーに従う

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

HDFS は追記型ファイルシステムのため、トランザクションは限定的。HadoopRecoverableWriter による hflush/hsync でデータ永続化を保証。

## パフォーマンス要件

- ブロックサイズに基づいたデータ局所性の活用
- 接続プールによるコネクション再利用
- 接続数制限による過負荷防止

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

- Kerberos 認証のサポート
- HDFS ACL によるアクセス制御
- データ転送の暗号化（dfs.encrypt.data.transfer）

## 備考

- viewfs:// スキームによるフェデレーテッド HDFS のサポート
- isDistributedFS() は常に true を返す
- RecoverableWriter は HadoopRecoverableWriter として実装

---

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

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

### 推奨読解順序

#### Step 1: ファクトリ実装を理解する

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | HadoopFsFactory.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactory.java` | ファクトリのメイン実装 |

**主要処理フロー**:
1. **57-60行目**: getScheme() で "*" を返却（任意のスキーム対応）
2. **63-66行目**: configure() で設定を保存
3. **69-198行目**: create() でファイルシステムを生成
4. **80-92行目**: Hadoop Configuration の読み込み
5. **96-107行目**: FileSystem.getFileSystemClass() でファイルシステムクラス取得
6. **120-163行目**: URI authority の解決
7. **167-176行目**: FileSystem の初期化
8. **181-185行目**: HadoopFileSystem でラップ
9. **208-226行目**: limitIfConfigured() で接続制限適用

**読解のコツ**: HadoopFsFactory は "*" スキームで登録され、任意の Hadoop 互換ファイルシステムをサポートする汎用ファクトリである。

#### Step 2: ファイルシステムラッパーを理解する

Flink FileSystem インターフェースへの適合を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | HadoopFileSystem.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java` | FileSystemラッパー |

**主要処理フロー**:
- **38行目**: Flink FileSystem を継承
- **55-57行目**: コンストラクタで Hadoop FileSystem を受け取る
- **73-75行目**: getWorkingDirectory()
- **82-84行目**: getUri()
- **125-136行目**: open() でファイルを開く
- **206-211行目**: createRecoverableWriter() で HadoopRecoverableWriter を生成
- **246-267行目**: getKindForScheme() でファイルシステム種別を判定

**読解のコツ**: HadoopFileSystem は Hadoop の FileSystem を Flink の FileSystem インターフェースに適合させるアダプタパターンの実装である。

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

チェックポイント対応の書き込み機能を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | HadoopRecoverableWriter.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriter.java` | RecoverableWriter実装 |

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

```
HadoopFsFactory.create(URI)
    │
    ├─ HadoopUtils.getHadoopConfiguration(flinkConfig)
    │      └─ Hadoop Configuration ロード
    │
    ├─ FileSystem.getFileSystemClass(scheme, hadoopConfig)
    │      └─ Hadoop SPI でファイルシステムクラス取得
    │
    ├─ fsClass.newInstance()
    │      └─ リフレクションでインスタンス生成
    │
    ├─ hadoopFs.initialize(initUri, hadoopConfig)
    │
    ├─ new HadoopFileSystem(hadoopFs)
    │      │
    │      └─ createRecoverableWriter()
    │             └─ new HadoopRecoverableWriter(fs)
    │
    └─ limitIfConfigured(fs, scheme, config)
           └─ new LimitedConnectionsFileSystem(fs, ...)
```

### データフロー図

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

URI (hdfs://namenode/...) ──▶ HadoopFsFactory ───────────▶ HadoopFileSystem
                                 │
Flink Configuration ──────────▶ HadoopUtils ────────────▶ Hadoop Configuration
                                 │
fs.defaultFS ─────────────────▶ URI解決 ─────────────────▶ 初期化URI
                                 │
Hadoop SPI ───────────────────▶ FileSystem.getFileSystemClass()
                                 │
接続制限設定 ─────────────────▶ LimitedConnectionsFileSystem
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| HadoopFsFactory.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactory.java` | ソース | ファクトリ |
| HadoopFileSystem.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java` | ソース | ファイルシステムラッパー |
| HadoopRecoverableWriter.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriter.java` | ソース | RecoverableWriter実装 |
| HadoopDataInputStream.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopDataInputStream.java` | ソース | 入力ストリームラッパー |
| HadoopDataOutputStream.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopDataOutputStream.java` | ソース | 出力ストリームラッパー |
| HadoopFileStatus.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileStatus.java` | ソース | ファイルステータスラッパー |
| HadoopBlockLocation.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopBlockLocation.java` | ソース | ブロック位置情報ラッパー |
| HadoopUtils.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/util/HadoopUtils.java` | ソース | Hadoopユーティリティ |
| HadoopConfigLoader.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/util/HadoopConfigLoader.java` | ソース | 設定ローダー |
| org.apache.flink.core.fs.FileSystemFactory | `flink-filesystems/flink-hadoop-fs/src/main/resources/META-INF/services/` | 設定 | SPIサービス登録 |
