# 機能設計書 55-Alibaba OSS

## 概要

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

### 本機能の処理概要

**業務上の目的・背景**：Alibaba Cloud（アリババクラウド）を利用する企業にとって、Object Storage Service（OSS）は主要なオブジェクトストレージサービスである。中国市場でのビッグデータ処理ワークロードにおいて、Flink ジョブの入出力データ、チェックポイント、セーブポイントの保存先として OSS を使用する需要に応えるため、本機能が提供される。

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

**主要な処理内容**：
1. OSSFileSystemFactory による FileSystem インスタンスの生成
2. Hadoop OSS ファイルシステム設定の読み込み
3. AliyunOSSFileSystem の初期化とラッピング
4. FlinkOSSFileSystem による HadoopFileSystem のラッピング
5. OSSRecoverableWriter によるチェックポイント対応書き込み（マルチパートアップロード）

**関連システム・外部連携**：
- Alibaba Cloud Object Storage Service（OSS）
- Hadoop Aliyun OSS コネクタ（hadoop-aliyun）

**権限による制御**：Alibaba Cloud AccessKey ID / AccessKey Secret による認証をサポート。

## 関連画面

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

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

## 機能種別

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

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| fs.oss.accessKeyId | String | Yes | Alibaba Cloud AccessKey ID | 非空文字列 |
| fs.oss.accessKeySecret | String | Yes | Alibaba Cloud AccessKey Secret | 非空文字列 |
| fs.oss.endpoint | String | Yes | OSSエンドポイント | 有効なURL |
| oss.upload.min.part.size | Long | No | マルチパートアップロードの最小パートサイズ | 10MB以上5GB以下 |
| oss.upload.max.concurrent.uploads | Integer | No | 最大同時アップロード数 | 正の整数（デフォルト: CPUコア数） |

### 入力データソース

- Flink Configuration（flink-conf.yaml）
- Hadoop Configuration

## 出力仕様

### 出力データ

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

### 出力先

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

## 処理フロー

### 処理シーケンス

```
1. FileSystemFactory.create(URI) が呼び出される
   └─ URIスキーム（oss://）を検証
2. getHadoopConfiguration() で設定を読み込み
   └─ fs.oss. プレフィックスの設定をマッピング
3. AliyunOSSFileSystem インスタンスを生成
4. AliyunOSSFileSystem.initialize(URI, Configuration) で初期化
5. OSSAccessor を生成
6. FlinkOSSFileSystem でラップして返却
   └─ RecoverableWriter サポート
```

### フローチャート

```mermaid
flowchart TD
    A[開始: create URI] --> B{URIスキーム確認}
    B -->|oss://| C[getHadoopConfiguration実行]
    B -->|その他| Z[エラー: 非対応スキーム]
    C --> D[fs.oss.*設定を読み込み]
    D --> E[AliyunOSSFileSystem生成]
    E --> F[initialize実行]
    F --> G{初期化成功?}
    G -->|Yes| H[OSSAccessor生成]
    G -->|No| Z2[エラー: 初期化失敗]
    H --> I[FlinkOSSFileSystem生成]
    I --> J[終了: FileSystem返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-55-01 | スキームマッピング | oss:// を AliyunOSSFileSystem にマッピング | 常時 |
| BR-55-02 | 設定プレフィックス | fs.oss. プレフィックスの設定を読み込み | 設定読み込み時 |
| BR-55-03 | 最小パートサイズ | マルチパートアップロードの最小パートサイズは10MB | 書き込み時 |
| BR-55-04 | 同時アップロード数 | デフォルトはCPUコア数 | RecoverableWriter使用時 |

### 計算ロジック

特になし

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | 接続エラー | OSSエンドポイントに接続できない | ネットワーク設定を確認 |
| OSSException | 認証エラー | AccessKey が無効 | AccessKey設定を確認 |
| IllegalArgumentException | 設定エラー | 必須設定が欠落またはパートサイズが範囲外 | flink-conf.yamlを確認 |

### リトライ仕様

Hadoop Aliyun OSS コネクタの標準リトライポリシーに従う

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

オブジェクトストレージのため、トランザクションは非対応。OSSRecoverableWriter によるマルチパートアップロードでアトミック書き込みを実現。

## パフォーマンス要件

- マルチパートアップロードによる並列書き込み
- 最小パートサイズ: 10MB（MULTIPART_UPLOAD_PART_SIZE_MIN）
- キャッシュスレッドプールによる非同期アップロード

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

- AccessKey は環境変数または安全な設定管理での提供を推奨
- 設定ファイルへの認証情報直接記載は非推奨
- OSS の IAM ポリシーで最小権限を設定

## 備考

- RecoverableWriter は OSSRecoverableWriter として実装
- 中国リージョン向けの最適化が施されている

---

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

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

### 推奨読解順序

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

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

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

**主要処理フロー**:
1. **47-50行目**: CONFIG_KEYS_TO_SHADE と FLINK_SHADING_PREFIX の定義
2. **56行目**: FLINK_CONFIG_PREFIXES = {"fs.oss."}
3. **58-75行目**: PART_UPLOAD_MIN_SIZE と MAX_CONCURRENT_UPLOADS のConfigOption定義
4. **78-80行目**: getScheme() で "oss" を返却
5. **83-86行目**: configure() で設定を保存
6. **89-115行目**: create() でファイルシステムを生成

**読解のコツ**: OSSFileSystemFactory は FileSystemFactory を実装し、ConfigOption でパートサイズと同時アップロード数を設定可能にしている。

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

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

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

**主要処理フロー**:
- **39行目**: HadoopFileSystem を継承
- **42行目**: MULTIPART_UPLOAD_PART_SIZE_MIN = 10MB
- **58-82行目**: コンストラクタで設定を受け取り、スレッドプールとファイルクリエイターを初期化
- **85-92行目**: createRecoverableWriter() で OSSRecoverableWriter を生成

**読解のコツ**: FlinkOSSFileSystem はキャッシュスレッドプールを使用して非同期でパートをアップロードする。RefCountedTmpFileCreator でローカル一時ファイルを管理。

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

OSS操作の実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | OSSAccessor.java | `flink-filesystems/flink-oss-fs-hadoop/src/main/java/org/apache/flink/fs/osshadoop/OSSAccessor.java` | OSS操作ヘルパー |

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

```
OSSFileSystemFactory.create(URI)
    │
    ├─ getHadoopConfiguration()
    │      └─ fs.oss.* 設定をHadoop Configurationにコピー
    │
    ├─ new AliyunOSSFileSystem()
    │
    ├─ AliyunOSSFileSystem.initialize(URI, hadoopConfig)
    │
    ├─ new OSSAccessor(fs)
    │
    └─ new FlinkOSSFileSystem(fs, partSize, maxUploads, tmpDir, accessor)
           │
           └─ createRecoverableWriter()
                  └─ new OSSRecoverableWriter(accessor, partSize, maxUploads, pool, fileCreator)
```

### データフロー図

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

URI (oss://bucket/path) ───▶ OSSFileSystemFactory ──────▶ FlinkOSSFileSystem
                                 │
Flink Configuration ──────────▶ getHadoopConfiguration() ▶ Hadoop Configuration
                                 │
OSS Credentials ──────────────▶ AliyunOSSFileSystem ────▶ OSS API Calls
                                 │
RecoverableWriter Request ────▶ OSSRecoverableWriter
                                 │
                             OSSRecoverableMultipartUpload
                                 │
                             OSSAccessor ───────────────▶ Multipart Upload API
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| OSSFileSystemFactory.java | `flink-filesystems/flink-oss-fs-hadoop/src/main/java/org/apache/flink/fs/osshadoop/OSSFileSystemFactory.java` | ソース | ファクトリ |
| FlinkOSSFileSystem.java | `flink-filesystems/flink-oss-fs-hadoop/src/main/java/org/apache/flink/fs/osshadoop/FlinkOSSFileSystem.java` | ソース | ファイルシステムラッパー |
| OSSAccessor.java | `flink-filesystems/flink-oss-fs-hadoop/src/main/java/org/apache/flink/fs/osshadoop/OSSAccessor.java` | ソース | OSS操作ヘルパー |
| OSSRecoverableWriter.java | `flink-filesystems/flink-oss-fs-hadoop/src/main/java/org/apache/flink/fs/osshadoop/writer/OSSRecoverableWriter.java` | ソース | RecoverableWriter実装 |
| OSSRecoverableMultipartUpload.java | `flink-filesystems/flink-oss-fs-hadoop/src/main/java/org/apache/flink/fs/osshadoop/writer/OSSRecoverableMultipartUpload.java` | ソース | マルチパートアップロード |
| OSSRecoverable.java | `flink-filesystems/flink-oss-fs-hadoop/src/main/java/org/apache/flink/fs/osshadoop/writer/OSSRecoverable.java` | ソース | リカバリ情報 |
| org.apache.flink.core.fs.FileSystemFactory | `flink-filesystems/flink-oss-fs-hadoop/src/main/resources/META-INF/services/` | 設定 | SPIサービス登録 |
