# Apache Flink コードリーディングガイドライン

## はじめに

このガイドラインは、Apache Flinkのコードベースを効率的に理解するための手引きです。
Javaに精通していないエンジニアでも、段階的に学習できるよう構成されています。

**対象読者:**
- プロジェクトに新規参画するエンジニア
- 他言語からの経験者
- コードレビューを行う担当者

---

## 1. 言語基礎

> このセクションでは、Java/Scalaの基本構文とFlink固有の概念を解説します。

### 1.1 プログラム構造

Apache Flinkは主にJava 11/17/21で実装されており、一部でScala 2.12も使用されています。Mavenによるマルチモジュール構成を採用しています。

**基本的なパッケージ構造:**
```java
// ファイル: flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:19
package org.apache.flink.runtime.jobmaster;
```

すべてのFlinkコードは `org.apache.flink` パッケージ配下に配置されています。

### 1.2 データ型と変数

Flinkでは、型安全性を重視した設計がなされています。主要なデータ型として以下が使用されます。

**TypeInformation による型情報の取得:**
```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:56
.map(word -> new WordWithCount(word, 1)).returns(TypeInformation.of(WordWithCount.class))
```

**Tuple型の使用:**
```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:23
import org.apache.flink.api.java.tuple.Tuple2;
```

### 1.3 制御構造

Flinkでは、ストリーム処理のフロー制御にDataStream APIを使用します。

**データフローの構築:**
```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:140-157
text.flatMap(new Tokenizer())
    .name("tokenizer")
    .keyBy(value -> value.f0);
```

### 1.4 関数/メソッド定義

ユーザー定義関数（UDF）はFlinkアプリケーションの中核です。

**FlatMapFunction の実装例:**
```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:191-206
public static final class Tokenizer
        implements FlatMapFunction<String, Tuple2<String, Integer>> {

    @Override
    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
        String[] tokens = value.toLowerCase().split("\\W+");
        for (String token : tokens) {
            if (token.length() > 0) {
                out.collect(new Tuple2<>(token, 1));
            }
        }
    }
}
```

### 1.5 モジュール/インポート

Flinkはマルチモジュール構成を採用しており、各機能がモジュールとして分離されています。

**主要モジュールのインポート:**
```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:20-36
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
```

---

## 2. プロジェクト固有の概念

> このセクションでは、Apache Flink特有の概念を解説します。

### 2.1 フレームワーク固有の概念

**StreamExecutionEnvironment:**
Flinkアプリケーションのエントリーポイントとなるクラスです。

```java
// ファイル: flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java:129-142
/**
 * The StreamExecutionEnvironment is the context in which a streaming program is executed.
 * A LocalStreamEnvironment will cause execution in the current JVM,
 * a RemoteStreamEnvironment will cause execution on a remote setup.
 */
@Public
public class StreamExecutionEnvironment implements AutoCloseable {
```

**RuntimeExecutionMode:**
STREAMINGモードとBATCHモードの切り替えが可能です。

```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:113
env.setRuntimeMode(params.getExecutionMode());
```

### 2.2 プロジェクト独自のパターン

**アノテーションによるAPI安定性の明示:**

Flinkでは、APIの安定性を以下のアノテーションで明示しています。

| アノテーション | 意味 | 安定性 |
|--------------|------|-------|
| `@Public` | 公開API。マイナーバージョン間で互換性維持 | 高 |
| `@PublicEvolving` | 公開APIだが、インターフェースが変更される可能性あり | 中 |
| `@Internal` | 内部API。開発者向け | 低 |
| `@Experimental` | 実験的機能 | 非常に低 |

```java
// ファイル: flink-annotations/src/main/java/org/apache/flink/annotation/Public.java:29-36
/**
 * Classes, methods and fields with this annotation are stable across minor releases (1.0, 1.1, 1.2).
 * Only major releases (1.0, 2.0, 3.0) can break interfaces with this annotation.
 */
@Public
public @interface Public {}
```

---

## 3. 命名規則

> このセクションでは、プロジェクト全体で使用される命名規則を解説します。

### 3.1 ファイル・ディレクトリ命名

| パターン | 意味 | 例 |
|---------|------|-----|
| `flink-*` | Flinkモジュール | `flink-runtime`, `flink-streaming-java` |
| `*Options.java` | 設定オプション定義 | `CheckpointingOptions.java` |
| `*Gateway.java` | RPCゲートウェイインターフェース | `JobMasterGateway.java` |
| `*Factory.java` | ファクトリークラス | `DeclarativeSlotPoolFactory.java` |
| `*Handler.java` | ハンドラー/リスナー | `FatalErrorHandler.java` |

### 3.2 クラス・関数・変数命名

| プレフィックス/サフィックス | 意味 | 例 |
|---------------------------|------|-----|
| `*Master` | マスターコンポーネント | `JobMaster` |
| `*Manager` | 管理コンポーネント | `ResourceManager`, `SlotManager` |
| `*Executor` | 実行コンポーネント | `TaskExecutor` |
| `*Coordinator` | 調整コンポーネント | `CheckpointCoordinator` |
| `*Service` | サービス層 | `HighAvailabilityServices` |
| `*Function` | ユーザー定義関数 | `FlatMapFunction`, `MapFunction` |
| `*Operator` | オペレーター | `StreamOperator`, `ProcessOperator` |
| `*Transformation` | 変換定義 | `OneInputTransformation` |

### 3.3 プログラム分類一覧

| 分類 | 説明 | 主要クラス |
|-----|------|----------|
| クラスターコンポーネント | クラスター管理 | `ClusterEntrypoint`, `ResourceManager`, `Dispatcher` |
| ジョブ管理 | ジョブの実行管理 | `JobMaster`, `SchedulerNG` |
| タスク実行 | タスクの実行 | `TaskExecutor`, `Task` |
| ストリーム処理API | ユーザー向けAPI | `DataStream`, `StreamExecutionEnvironment` |
| テーブルAPI | SQL/Table API | `TableEnvironment`, `Table` |

---

## 4. ディレクトリ構造

> このセクションでは、プロジェクトのディレクトリ構造を解説します。

```
flink-master/
├── flink-annotations/          # アノテーション定義
├── flink-core-api/             # コアAPI定義
├── flink-core/                 # コア機能（型、設定、ファイルシステム）
├── flink-runtime/              # ランタイム（実行エンジン）
├── flink-streaming-java/       # DataStream API実装
├── flink-datastream-api/       # DataStream API定義
├── flink-table/                # Table/SQL API
├── flink-connectors/           # コネクター（ソース/シンク）
├── flink-formats/              # データフォーマット
├── flink-state-backends/       # 状態バックエンド
├── flink-clients/              # クライアントAPI
├── flink-rpc/                  # RPC層
├── flink-metrics/              # メトリクス
├── flink-filesystems/          # ファイルシステム
├── flink-kubernetes/           # Kubernetes連携
├── flink-yarn/                 # YARN連携
├── flink-examples/             # サンプルコード
├── flink-tests/                # テストスイート
├── flink-end-to-end-tests/     # E2Eテスト
├── flink-docs/                 # ドキュメント
├── tools/                      # ビルド/CI用ツール
└── pom.xml                     # 親POM
```

### 各ディレクトリの役割

| ディレクトリ | 役割 | 主要ファイル |
|-------------|------|-------------|
| `flink-annotations` | API安定性を示すアノテーション | `Public.java`, `Internal.java` |
| `flink-core` | コア機能（型システム、設定、ユーティリティ） | `Configuration.java`, `TypeInformation.java` |
| `flink-runtime` | 実行エンジン | `JobMaster.java`, `TaskExecutor.java`, `ResourceManager.java` |
| `flink-streaming-java` | DataStream API実装 | `DataStream.java`, `StreamExecutionEnvironment.java` |
| `flink-table` | Table/SQL API | `TableEnvironment.java` |
| `flink-connectors` | 外部システム接続 | `FileSource.java`, `FileSink.java` |
| `flink-state-backends` | 状態管理バックエンド | `RocksDBStateBackend.java` |
| `flink-examples` | サンプルコード | `WordCount.java` |

---

## 5. アーキテクチャ

> このセクションでは、プロジェクトのアーキテクチャパターンを解説します。

### 5.1 全体アーキテクチャ

Apache Flinkは、マスター・ワーカー型の分散アーキテクチャを採用しています。

```
┌──────────────────────────────────────────────────────────────────┐
│                        Flink Cluster                             │
├──────────────────────────────────────────────────────────────────┤
│  ┌──────────────────────────────────────────────────────────┐   │
│  │                   JobManager (Master)                     │   │
│  │  ┌─────────────┐ ┌─────────────┐ ┌───────────────────┐   │   │
│  │  │ Dispatcher  │ │ResourceMgr │ │     JobMaster     │   │   │
│  │  └─────────────┘ └─────────────┘ └───────────────────┘   │   │
│  │                                    ┌─────────────────┐   │   │
│  │                                    │   Scheduler     │   │   │
│  │                                    └─────────────────┘   │   │
│  └──────────────────────────────────────────────────────────┘   │
│                              │                                   │
│                       RPC Communication                          │
│                              │                                   │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │                 TaskManagers (Workers)                    │   │
│  │  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐      │   │
│  │  │TaskExecutor 1│ │TaskExecutor 2│ │TaskExecutor N│      │   │
│  │  │ ┌──────────┐ │ │ ┌──────────┐ │ │ ┌──────────┐ │      │   │
│  │  │ │  Task    │ │ │ │  Task    │ │ │ │  Task    │ │      │   │
│  │  │ └──────────┘ │ │ └──────────┘ │ │ └──────────┘ │      │   │
│  │  └──────────────┘ └──────────────┘ └──────────────┘      │   │
│  └──────────────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────────┘
```

### 5.2 レイヤー構成

| レイヤー | 責務 | 代表的なファイル |
|---------|------|-----------------|
| API Layer | ユーザー向けAPI | `StreamExecutionEnvironment.java`, `DataStream.java` |
| Graph Layer | ジョブグラフの構築 | `StreamGraph.java`, `JobGraph.java` |
| Runtime Layer | 実行管理 | `JobMaster.java`, `Scheduler.java` |
| Execution Layer | タスク実行 | `TaskExecutor.java`, `Task.java` |
| State Layer | 状態管理 | `StateBackend.java`, `CheckpointCoordinator.java` |

### 5.3 データフロー

```
ユーザーコード (DataStream API)
         │
         ▼
StreamGraph (論理実行グラフ)
         │
         ▼
JobGraph (物理実行グラフ)
         │
         ▼
ExecutionGraph (実行時グラフ)
         │
         ▼
Task (分散タスク)
```

1. **ユーザーコード**: DataStream APIを使用してパイプラインを定義
2. **StreamGraph**: 論理的な実行グラフを生成
3. **JobGraph**: 最適化された物理実行グラフに変換
4. **ExecutionGraph**: 実行時のスケジューリング用グラフ
5. **Task**: TaskManagerで実行される実際のタスク

---

## 6. 主要コンポーネント

> このセクションでは、主要なコンポーネントとその連携を解説します。

### 6.1 エントリーポイント

**アプリケーション実行のエントリーポイント:**

```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:78-83
public static void main(String[] args) throws Exception {
    final CLI params = CLI.fromArgs(args);
    // Create the execution environment. This is the main entrypoint
    // to building a Flink application.
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
```

**クラスターのエントリーポイント:**

```java
// ファイル: flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java:19
package org.apache.flink.runtime.entrypoint;
```

ClusterEntrypointは、Flinkクラスターの起動点となる抽象クラスです。

### 6.2 ビジネスロジック

**JobMaster - ジョブの実行管理:**

```java
// ファイル: flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:19
package org.apache.flink.runtime.jobmaster;
```

JobMasterは個々のジョブを管理し、タスクのスケジューリング、チェックポイントの調整、フェイルオーバーを担当します。

**Dispatcher - ジョブの受付と管理:**

```java
// ファイル: flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:19
package org.apache.flink.runtime.dispatcher;
```

Dispatcherはジョブの提出を受け付け、JobMasterを起動します。

### 6.3 データアクセス

**ステートバックエンド:**

Flinkは複数の状態バックエンドをサポートしています。

| バックエンド | 用途 | モジュール |
|------------|------|----------|
| HashMapStateBackend | インメモリ状態管理 | flink-runtime |
| RocksDBStateBackend | 大規模状態管理 | flink-statebackend-rocksdb |
| ForstStateBackend | 非同期状態アクセス | flink-statebackend-forst |

**チェックポイント:**

```java
// ファイル: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
// CheckpointCoordinatorは、分散スナップショットを調整し、
// 障害発生時にアプリケーション状態を復元します。
```

### 6.4 ユーティリティ/共通機能

**設定オプション:**

```java
// ファイル: flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
// Configurationクラスは、Flinkアプリケーションの設定を管理します。
```

---

## 7. よく使われるパターン

> このセクションでは、コード内で頻出するパターンを解説します。

### パターン一覧

| パターン | 説明 | 出現頻度 | 代表的なファイル |
|---------|------|---------|-----------------|
| Builder パターン | 複雑なオブジェクトの構築 | 高 | `FileSink.builder()` |
| Factory パターン | オブジェクト生成の抽象化 | 高 | `DeclarativeSlotPoolFactory` |
| Gateway/RPC パターン | 分散通信 | 高 | `JobMasterGateway` |
| Function パターン | ユーザー定義ロジック | 高 | `FlatMapFunction` |

### 各パターンの詳細

#### パターン1: DataStream変換チェーン

**目的:** ストリーム処理パイプラインの構築

**実装例:**
```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:136-157
KeyedStream<Tuple2<String, Integer>, String> keyedStream =
        text.flatMap(new Tokenizer())
                .name("tokenizer")
                .keyBy(value -> value.f0);
DataStream<Tuple2<String, Integer>> counts =
        keyedStream
                .sum(1)
                .name("counter");
```

**解説:** DataStream APIでは、メソッドチェーンを使用して変換を連結します。各変換は新しいDataStreamを返し、遅延評価されます。

#### パターン2: Sink/Source Builder

**目的:** データの入出力定義

**実装例:**
```java
// ファイル: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java:163-172
counts.sinkTo(
        FileSink.<Tuple2<String, Integer>>forRowFormat(
                        params.getOutput().get(), new SimpleStringEncoder<>())
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withMaxPartSize(MemorySize.ofMebiBytes(1))
                                .withRolloverInterval(Duration.ofSeconds(10))
                                .build())
                .build())
```

---

## 8. 業務フロー追跡の実践例

> このセクションでは、実際の業務フローをコードで追跡する方法を解説します。

### 8.1 フロー追跡の基本手順

1. エントリーポイントを特定（main関数またはStreamExecutionEnvironment）
2. 処理の流れを追跡（DataStream変換チェーン）
3. データの変換を確認（map, flatMap, keyBy等）
4. 最終的な出力を確認（sinkTo, print）

### 8.2 フロー追跡の実例

#### 例1: WordCountジョブの実行フロー

**概要:** テキストファイルから単語をカウントするストリーム処理

**処理フロー:**
```
main() → StreamExecutionEnvironment → FileSource → flatMap → keyBy → sum → FileSink
```

**詳細な追跡:**

1. **環境の取得** (`WordCount.java:83`)
   ```java
   final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   ```

2. **データソースの定義** (`WordCount.java:123-131`)
   ```java
   FileSource.FileSourceBuilder<String> builder =
           FileSource.forRecordStreamFormat(
                   new TextLineInputFormat(), params.getInputs().get());
   text = env.fromSource(builder.build(), WatermarkStrategy.noWatermarks(), "file-input");
   ```

3. **変換の適用** (`WordCount.java:140-157`)
   ```java
   text.flatMap(new Tokenizer())
       .keyBy(value -> value.f0)
       .sum(1);
   ```

4. **シンクへの出力** (`WordCount.java:163-172`)
   ```java
   counts.sinkTo(FileSink...);
   ```

5. **ジョブの実行** (`WordCount.java:179`)
   ```java
   env.execute("WordCount");
   ```

### 8.3 フロー追跡チェックリスト

- [x] エントリーポイントを特定したか（main関数、StreamExecutionEnvironment）
- [x] 呼び出し関係を把握したか（DataStream変換チェーン）
- [x] データの変換ポイントを確認したか（flatMap, keyBy, sum）
- [x] エラーハンドリングを確認したか（例外処理、フェイルオーバー）
- [x] 最終的な出力を確認したか（FileSink, print）

---

## 9. 設計書の参照順序

> このセクションでは、プロジェクト理解のための設計書参照順序を案内します。

### 9.1 目的別ロードマップ

#### 全体像を把握したい場合
1. `README.md` - プロジェクト概要
2. `flink-examples/` - サンプルコードで動作を理解
3. `flink-docs/` - 公式ドキュメント

#### DataStream APIを理解したい場合
1. `flink-examples/flink-examples-streaming/` - ストリーム処理サンプル
2. `flink-runtime/.../streaming/api/` - DataStream API実装
3. `flink-streaming-java/` - ストリーム処理実装

#### ランタイムを理解したい場合
1. `flink-runtime/.../entrypoint/` - クラスターエントリーポイント
2. `flink-runtime/.../jobmaster/` - ジョブ管理
3. `flink-runtime/.../taskexecutor/` - タスク実行
4. `flink-runtime/.../scheduler/` - スケジューリング

#### 改修作業を行う場合
1. 対象機能のサンプルコード（`flink-examples/`）
2. 対象モジュールのテストコード（`**/src/test/`）
3. 対象モジュールの実装コード

### 9.2 ドキュメント一覧

| ドキュメント | 概要 | 参照タイミング |
|-------------|------|---------------|
| `README.md` | プロジェクト概要、ビルド方法 | 最初に参照 |
| `flink-docs/` | 公式ドキュメント | 機能理解時 |
| `flink-examples/` | サンプルコード | 実装参考時 |
| `flink-architecture-tests/` | アーキテクチャテスト | 設計確認時 |

---

## 10. トラブルシューティング

> このセクションでは、コードリーディング時によくある問題と解決法を解説します。

### よくある疑問と回答

#### Q: StreamGraphとJobGraphの違いは何ですか？
A: StreamGraphは論理的な実行グラフで、ユーザーのDataStream APIから生成されます。JobGraphは最適化された物理実行グラフで、JobManagerに送信されます。参照: `flink-runtime/.../jobgraph/JobGraph.java`

#### Q: @Publicと@PublicEvolvingの使い分けは？
A: @Publicはマイナーバージョン間で互換性が保証されます。@PublicEvolvingは公開APIですが、インターフェースが変更される可能性があります。参照: `flink-annotations/`

#### Q: チェックポイントとセーブポイントの違いは？
A: チェックポイントは自動的な障害回復用のスナップショットです。セーブポイントはユーザーが手動でトリガーするスナップショットで、ジョブのアップグレードや移行に使用します。参照: `flink-runtime/.../checkpoint/`

### デバッグのヒント

1. **ログレベルの調整**: `log4j2.properties`でログレベルを変更
2. **Web UI**: `http://localhost:8081`でジョブの状態を確認
3. **メトリクス**: Prometheusなどでメトリクスを監視
4. **FlameGraph**: CPUプロファイリングでボトルネックを特定

---

## 付録

### A. 用語集

| 用語 | 説明 |
|-----|------|
| DataStream | 型付けされたデータストリーム |
| Transformation | データ変換操作 |
| Operator | 変換を実行するコンポーネント |
| Task | TaskManagerで実行される処理単位 |
| Subtask | Taskの並列インスタンス |
| JobGraph | 最適化された実行グラフ |
| ExecutionGraph | 実行時のスケジューリング用グラフ |
| Checkpoint | 障害回復用のスナップショット |
| Savepoint | ユーザートリガーのスナップショット |
| State Backend | 状態を保存するストレージ |
| Watermark | イベント時間の進行を示すマーカー |

### B. ファイル一覧

| ファイル/ディレクトリ | 説明 | 主な内容 |
|---------------------|------|---------|
| `pom.xml` | 親POM | モジュール定義、依存関係 |
| `flink-core/` | コアモジュール | 型、設定、ユーティリティ |
| `flink-runtime/` | ランタイムモジュール | 実行エンジン |
| `flink-streaming-java/` | ストリーミングモジュール | DataStream実装 |
| `flink-table/` | テーブルモジュール | Table/SQL API |
| `flink-connectors/` | コネクター | ソース/シンク |
| `flink-state-backends/` | 状態バックエンド | RocksDB, ForSt |
| `flink-examples/` | サンプル | WordCount等 |

### C. 参考資料

- [Apache Flink公式ドキュメント](https://flink.apache.org/docs/)
- [Flink GitHub](https://github.com/apache/flink)
- [Flink Improvement Proposals (FLIPs)](https://cwiki.apache.org/confluence/display/FLINK/Flink+Improvement+Proposals)
- [Flink Mailing Lists](https://flink.apache.org/community.html#mailing-lists)
