# 機能設計書 54-Google Cloud Storage

## 概要

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

### 本機能の処理概要

**業務上の目的・背景**：Google Cloud Platform（GCP）を利用する企業にとって、Google Cloud Storage は主要なオブジェクトストレージサービスである。Flink ジョブの入出力データ、チェックポイント、セーブポイントの保存先として GCS を使用する需要に応えるため、本機能が提供される。GCS は高い耐久性と可用性を持ち、ビッグデータワークロードに適している。

**機能の利用シーン**：
- ストリーム処理ジョブのチェックポイントを GCS に保存する場合
- バッチ処理ジョブの入力データを GCS から読み込む場合
- 処理結果を GCS バケットに出力する場合
- GCP 環境でのデータレイク構築

**主要な処理内容**：
1. GSFileSystemFactory による FileSystem インスタンスの生成
2. Google Cloud Storage API との連携設定
3. Hadoop GCS コネクタ（GoogleHadoopFileSystem）の初期化
4. GSFileSystem による HadoopFileSystem のラッピング
5. GSRecoverableWriter によるチェックポイント対応書き込み

**関連システム・外部連携**：
- Google Cloud Storage
- Google Cloud Storage API (JSON)
- Hadoop GCS コネクタ（gcs-connector）

**権限による制御**：Google Cloud サービスアカウントキーまたは GCE メタデータサービスによる認証をサポート。

## 関連画面

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

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

## 機能種別

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

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| google.cloud.auth.service.account.json.keyfile | String | No（GCE外での実行時に推奨） | サービスアカウントキーファイルパス | 有効なファイルパス |
| gs.http.connect-timeout | Integer | No | HTTP接続タイムアウト（ミリ秒） | 正の整数 |
| gs.http.read-timeout | Integer | No | HTTP読み取りタイムアウト（ミリ秒） | 正の整数 |
| gs.writer.chunk.size | Integer | No | 書き込みチャンクサイズ | 正の整数 |

### 入力データソース

- Flink Configuration（flink-conf.yaml）
- 環境変数（GOOGLE_APPLICATION_CREDENTIALS）
- Hadoop Configuration
- GCE メタデータサービス（GCE上で実行時）

## 出力仕様

### 出力データ

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

### 出力先

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

## 処理フロー

### 処理シーケンス

```
1. FileSystemFactory.create(URI) が呼び出される
   └─ URIスキーム（gs://）を検証
2. configure() で設定を読み込み
   └─ Hadoop設定とGSFileSystemOptionsの構築
3. Google Storage クライアントの初期化
   └─ 認証情報、リトライ設定、HTTPオプションの適用
4. GoogleHadoopFileSystem インスタンスを生成・初期化
5. GSBlobStorageImpl でストレージ操作をラップ
6. GSFileSystem でラップして返却
   └─ RecoverableWriter サポート
```

### フローチャート

```mermaid
flowchart TD
    A[開始: create URI] --> B{URIスキーム確認}
    B -->|gs://| C[configure実行]
    B -->|その他| Z[エラー: 非対応スキーム]
    C --> D[Hadoop Configuration構築]
    D --> E[GSFileSystemOptions構築]
    E --> F[Google Storage クライアント初期化]
    F --> G[認証情報取得]
    G --> H[GoogleHadoopFileSystem生成]
    H --> I[GoogleHadoopFileSystem.initialize]
    I --> J{初期化成功?}
    J -->|Yes| K[GSFileSystem生成]
    J -->|No| Z2[エラー: 初期化失敗]
    K --> L[終了: FileSystem返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-54-01 | スキームマッピング | gs:// を GoogleHadoopFileSystem にマッピング | 常時 |
| BR-54-02 | 認証優先順位 | 明示的設定 > 環境変数 > GCE メタデータ | 認証情報解決時 |
| BR-54-03 | チャンク書き込み | RecoverableWriter は指定チャンクサイズで分割書き込み | 大容量ファイル書き込み時 |
| BR-54-04 | CRC32Cチェックサム | データ整合性検証にCRC32Cを使用 | 書き込み時 |

### 計算ロジック

特になし

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | 接続エラー | GCSエンドポイントに接続できない | ネットワーク設定を確認 |
| GoogleJsonResponseException | 認証エラー | 認証情報が無効または権限不足 | サービスアカウント設定を確認 |
| RuntimeException | 初期化エラー | GoogleHadoopFileSystem の初期化失敗 | 設定を確認 |

### リトライ仕様

GSFileSystemOptions で設定可能なリトライ設定：
- maxAttempts: 最大リトライ回数
- initialRpcTimeout: 初期RPCタイムアウト
- rpcTimeoutMultiplier: タイムアウト乗数
- maxRpcTimeout: 最大RPCタイムアウト
- totalTimeout: 総タイムアウト

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

オブジェクトストレージのため、トランザクションは非対応。GSRecoverableWriter によるコンポーズ操作でアトミック書き込みを実現。

## パフォーマンス要件

- チャンク単位での並列アップロード
- HTTP接続プールの再利用
- 適切なチャンクサイズ設定により性能最適化

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

- サービスアカウントキーは安全な場所に保管
- 最小権限の原則でIAMロールを設定
- VPC Service Controls の活用を推奨

## 備考

- GSRecoverableWriter は GCS の compose API を活用してチェックポイントをサポート
- Hadoop GCS コネクタとの互換性を維持

---

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

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

### 推奨読解順序

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | GSFileSystemFactory.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemFactory.java` | ファクトリのメイン実装 |
| 1-2 | GSFileSystemOptions.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemOptions.java` | オプション設定クラス |

**主要処理フロー**:
1. **55行目**: SCHEME = "gs" の定義
2. **78-79行目**: コンストラクタでログ出力
3. **83-110行目**: configure() で設定を読み込み、Storage クライアントを初期化
4. **112-119行目**: getHttpTransportOptions() でHTTPオプション構築
5. **122-138行目**: getRetrySettings() でリトライ設定構築
6. **146-161行目**: create() でファイルシステムを生成

**読解のコツ**: GSFileSystemFactory は FileSystemFactory を実装し、Google Cloud Storage API クライアントを初期化する。Storage オブジェクトはスレッドセーフで、全ファイルシステムインスタンス間で共有される。

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

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

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

**主要処理フロー**:
- **33行目**: HadoopFileSystem を継承
- **41-48行目**: コンストラクタで GoogleHadoopFileSystem, Storage, オプションを受け取る
- **51-58行目**: createRecoverableWriter() で GSRecoverableWriter を生成

**読解のコツ**: GSFileSystem は HadoopFileSystem を継承しつつ、RecoverableWriter として独自の GSRecoverableWriter を提供する。

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | GSRecoverableWriter.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/writer/GSRecoverableWriter.java` | RecoverableWriter実装 |
| 3-2 | GSBlobStorageImpl.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/storage/GSBlobStorageImpl.java` | Blob Storage操作 |

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

```
GSFileSystemFactory.create(URI)
    │
    ├─ configure(Configuration)
    │      ├─ ConfigUtils.getHadoopConfiguration()
    │      ├─ new GSFileSystemOptions(flinkConfig)
    │      ├─ getHttpTransportOptions()
    │      ├─ getRetrySettings()
    │      └─ StorageOptions.newBuilder().build().getService()
    │
    ├─ new GoogleHadoopFileSystem()
    │
    ├─ GoogleHadoopFileSystem.initialize(URI, hadoopConfig)
    │
    └─ new GSFileSystem(googleHadoopFileSystem, storage, fileSystemOptions)
           │
           └─ createRecoverableWriter()
                  ├─ new GSBlobStorageImpl(storage)
                  └─ new GSRecoverableWriter(blobStorage, fileSystemOptions)
```

### データフロー図

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

URI (gs://bucket/path) ────▶ GSFileSystemFactory ───────▶ GSFileSystem
                                 │
Flink Configuration ──────────▶ ConfigUtils ────────────▶ Hadoop Configuration
                                 │
                             GSFileSystemOptions ───────▶ HTTP/リトライ設定
                                 │
Google Credentials ───────────▶ StorageOptions ─────────▶ Storage クライアント
                                 │
RecoverableWriter Request ────▶ GSRecoverableWriter
                                 │
                             GSBlobStorageImpl ─────────▶ GCS compose API
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| GSFileSystemFactory.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemFactory.java` | ソース | ファクトリ |
| GSFileSystem.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystem.java` | ソース | ファイルシステムラッパー |
| GSFileSystemOptions.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemOptions.java` | ソース | オプション設定 |
| GSRecoverableWriter.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/writer/GSRecoverableWriter.java` | ソース | RecoverableWriter実装 |
| GSBlobStorageImpl.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/storage/GSBlobStorageImpl.java` | ソース | Blob Storage操作 |
| GSBlobStorage.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/storage/GSBlobStorage.java` | ソース | Blob Storageインターフェース |
| ConfigUtils.java | `flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/utils/ConfigUtils.java` | ソース | 設定ユーティリティ |
| org.apache.flink.core.fs.FileSystemFactory | `flink-filesystems/flink-gs-fs-hadoop/src/main/resources/META-INF/services/` | 設定 | SPIサービス登録 |
