# 機能設計書 40-Hadoop互換性

## 概要

本ドキュメントは、Apache FlinkのHadoop互換性機能について記載する。この機能は、Hadoop InputFormat/OutputFormatをFlinkで利用可能にするラッパーを提供し、既存のHadoopエコシステムとの連携を実現する。

### 本機能の処理概要

Hadoop互換性モジュールは、Hadoop MapReduce APIのInputFormat/OutputFormatをFlinkのInputFormat/OutputFormatとして使用できるようにラップする。mapred（旧API）とmapreduce（新API）の両方をサポートし、Key-ValueペアをFlink Tuple2に変換する。

**業務上の目的・背景**：既存のHadoopエコシステムには多くのInputFormat/OutputFormat実装が存在する。Hadoop互換性モジュールを使用することで、これらの既存実装をFlinkジョブで再利用でき、移行コストを削減できる。

**機能の利用シーン**：
- 既存Hadoop InputFormatの再利用
- HDFS上のファイル読み取り
- SequenceFileの読み書き
- カスタムHadoop InputFormat/OutputFormatの利用

**主要な処理内容**：
1. Hadoop InputFormatのラップ
2. Hadoop OutputFormatのラップ
3. InputSplitの変換
4. Key-Value → Tuple2変換
5. JobConf設定の管理

**関連システム・外部連携**：
- Hadoop InputFormat/OutputFormat
- HDFS/その他Hadoopファイルシステム
- Hadoop JobConf

**権限による制御**：Hadoopの認証・認可（Kerberos等）に依存。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は画面とは直接関連しない（コネクタ処理） |

## 機能種別

データ入出力 / Hadoop連携 / 互換性レイヤー

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| mapredInputFormat | InputFormat<K,V> | Yes | Hadoop InputFormat | 非null |
| key | Class<K> | Yes | キークラス | 非null |
| value | Class<V> | Yes | 値クラス | 非null |
| job | JobConf/Job | No | Hadoop設定 | デフォルトJobConf |
| inputPath | String | File時Yes | 入力パス | 有効なパス |

### 入力データソース

- Hadoop InputFormatでサポートされるデータソース
- HDFS、ローカルファイルシステム、S3等

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Tuple2<K, V> | Tuple2 | Key-Valueペア |

### 出力先

- Flink DataSet/DataStream

## 処理フロー

### 処理シーケンス

```
1. HadoopInputFormat作成
   └─ Hadoop InputFormat、Key/Valueクラス、JobConf設定

2. configure()
   └─ Configurable/JobConfigurableの設定

3. createInputSplits()
   └─ Hadoop InputSplitをHadoopInputSplitにラップ

4. open()
   └─ RecordReader取得、Key/Value初期化

5. nextRecord()
   └─ RecordReader.next()でKey-Value取得
   └─ Tuple2に変換

6. close()
   └─ RecordReaderクローズ
```

### フローチャート

```mermaid
flowchart TD
    A[HadoopInputFormat作成] --> B[JobConf設定]
    B --> C[configure]
    C --> D[createInputSplits]
    D --> E[HadoopInputSplitラップ]
    E --> F[open]
    F --> G[RecordReader取得]
    G --> H[nextRecord]
    H --> I[recordReader.next]
    I --> J[Key-Value取得]
    J --> K[Tuple2変換]
    K --> L{reachedEnd?}
    L -->|No| H
    L -->|Yes| M[close]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-40-01 | mapred/mapreduce両対応 | 旧API・新API両方をサポート | 常時 |
| BR-40-02 | Tuple2変換 | Key-ValueをTuple2<K,V>に変換 | 常時 |
| BR-40-03 | 排他制御 | open/configure/closeは排他制御 | マルチスレッド時 |
| BR-40-04 | Credentials継承 | Hadoop Credentialsを継承 | セキュリティ有効時 |

### 計算ロジック

出力型: `TupleTypeInfo<Tuple2<K, V>>`
- f0: Key（TypeExtractor.createTypeInfo(keyClass)）
- f1: Value（TypeExtractor.createTypeInfo(valueClass)）

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

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

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

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

該当なし（Hadoop InputFormat依存）

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IO | IOException | InputFormat読み取り失敗 | Hadoop設定確認 |
| CLASS | RuntimeException | クラスロード失敗 | クラスパス確認 |
| AUTH | Exception | Kerberos認証失敗 | 認証設定確認 |

### リトライ仕様

Flink標準のリトライメカニズムに依存

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

Hadoop InputFormat/OutputFormatのトランザクション特性に依存

## パフォーマンス要件

- 排他制御（OPEN_MUTEX、CONFIGURE_MUTEX、CLOSE_MUTEX）によるオーバーヘッド
- Hadoop JVM分離前提のため、マルチスレッド時に注意

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

- Hadoop Credentialsの継承
- UserGroupInformationからのCredentials取得

## 備考

- @Public（mapred HadoopInputFormat）
- @Internal（HadoopInputFormatBase）
- HadoopInputsユーティリティクラスで簡易利用可能

---

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

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

### 推奨読解順序

#### Step 1: HadoopInputsユーティリティを理解する

ユーティリティクラスでAPIの使い方を把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | HadoopInputs.java | `flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/` | ユーティリティAPI |

**読解のコツ**:
- **29-39行目**: クラスコメントで設計意図を把握
- **50-61行目**: readHadoopFile()（mapred FileInputFormat）
- **83-90行目**: readSequenceFile()
- **98-104行目**: createHadoopInput()（mapred InputFormat）
- **151-159行目**: createHadoopInput()（mapreduce InputFormat）

#### Step 2: HadoopInputFormatを理解する

メインのラッパークラスを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | HadoopInputFormat.java | `flink-hadoop-compatibility/.../mapred/` | mapred版ラッパー |

**読解のコツ**:
- **32-39行目**: クラスコメント
- **41-42行目**: HadoopInputFormatBase継承、ResultTypeQueryable実装
- **46-52行目**: コンストラクタ
- **61-73行目**: nextRecord()でTuple2変換
- **75-79行目**: getProducedType()で型情報取得

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

基底クラスの実装詳細を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | HadoopInputFormatBase.java | 同パス | 共通基底クラス |

**読解のコツ**:
- **66-72行目**: 排他制御用ミューテックス
- **86-98行目**: コンストラクタ、HadoopUtils.mergeHadoopConf()
- **108-120行目**: configure()（Configurable/JobConfigurable対応）
- **154-163行目**: createInputSplits()
- **170-185行目**: open()（RecordReader取得）
- **187-198行目**: reachedEnd()、fetchNext()

#### Step 4: mapreduce版を理解する

新API版の実装を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | HadoopInputFormat.java | `flink-hadoop-compatibility/.../mapreduce/` | mapreduce版ラッパー |

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

```
HadoopInputs（ユーティリティ）
    │
    ├─ readHadoopFile(FileInputFormat, key, value, path, job)
    │      └─ createHadoopInput()
    │
    ├─ readSequenceFile(key, value, path)
    │      └─ readHadoopFile(SequenceFileInputFormat)
    │
    └─ createHadoopInput(InputFormat, key, value, job)
           └─ new HadoopInputFormat(...)

HadoopInputFormat
    │
    ├─ extends HadoopInputFormatBase
    │      ├─ configure() [synchronized CONFIGURE_MUTEX]
    │      ├─ createInputSplits()
    │      │      └─ mapredInputFormat.getSplits()
    │      │      └─ HadoopInputSplit作成
    │      ├─ open() [synchronized OPEN_MUTEX]
    │      │      └─ mapredInputFormat.getRecordReader()
    │      ├─ reachedEnd()
    │      │      └─ fetchNext()
    │      │             └─ recordReader.next(key, value)
    │      └─ close() [synchronized CLOSE_MUTEX]
    │             └─ recordReader.close()
    │
    └─ nextRecord(record)
           └─ record.f0 = key
           └─ record.f1 = value
```

### データフロー図

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

Hadoop InputFormat ────▶ HadoopInputFormat ──────────▶ DataSet/DataStream
                               │
                               ├─ createInputSplits()
                               │      └─ Hadoop InputSplit → HadoopInputSplit
                               │
                               ├─ open(split)
                               │      └─ RecordReader<K, V>
                               │
                               └─ nextRecord()
                                      ├─ key
                                      ├─ value
                                      └─ Tuple2<K, V>

mapred API:
org.apache.hadoop.mapred.InputFormat ──▶ HadoopInputFormat (mapred)

mapreduce API:
org.apache.hadoop.mapreduce.InputFormat ──▶ HadoopInputFormat (mapreduce)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| HadoopInputs.java | `flink-hadoop-compatibility/.../hadoopcompatibility/` | ユーティリティ | ファクトリメソッド |
| HadoopInputFormat.java | `.../mapred/` | ラッパー | mapred版InputFormat |
| HadoopInputFormatBase.java | `.../mapred/` | 基底クラス | 共通実装 |
| HadoopOutputFormat.java | `.../mapred/` | ラッパー | mapred版OutputFormat |
| HadoopInputFormat.java | `.../mapreduce/` | ラッパー | mapreduce版InputFormat |
| HadoopOutputFormat.java | `.../mapreduce/` | ラッパー | mapreduce版OutputFormat |
| HadoopInputSplit.java | `.../mapred/wrapper/` | ラッパー | InputSplit |
| HadoopDummyReporter.java | `.../mapred/wrapper/` | ダミー | Reporter実装 |
| WritableTypeInfo.java | `.../typeutils/` | 型情報 | Writable型情報 |
| WritableSerializer.java | `.../typeutils/runtime/` | シリアライザ | Writableシリアライズ |
| HadoopUtils.java | `.../mapred/utils/` | ユーティリティ | Hadoop設定マージ |
