# 機能設計書 39-チェックポイント管理

## 概要

本ドキュメントは、Apache SparkのStructured Streamingにおけるチェックポイント管理機能に関する機能設計書である。ストリーミングクエリの進行状況（オフセット等）を永続ストレージに保存し、障害復旧時のExactly-Once処理を保証するチェックポイント機構を詳細に記述する。

### 本機能の処理概要

**業務上の目的・背景**：ストリーミング処理では、アプリケーション障害やインフラ障害からの復旧時に、処理済みデータの重複処理やデータ損失を防ぐ必要がある。チェックポイント管理は、各バッチの処理進行状況（ソースオフセット、コミット完了情報）を永続ストレージに記録し、障害復旧時に正確な再開ポイントを提供することで、Exactly-Once処理保証を実現する。

**機能の利用シーン**：ストリーミングクエリの障害復旧、クエリのバージョンアップ時の状態継続、クラスタ移行時のクエリ再開。

**主要な処理内容**：
1. OffsetSeqLog：各バッチのソースオフセットを記録するメタデータログ
2. CommitLog：各バッチの完了（コミット）状態を記録するメタデータログ
3. HDFSMetadataLog：HDFS互換ファイルシステム上のメタデータログ基盤
4. CheckpointFileManager：チェックポイントファイルのアトミックな書き込み・リネーム管理
5. AsyncOffsetSeqLog/AsyncCommitLog：非同期書き込みに対応したログ実装
6. StreamingQueryCheckpointMetadata：クエリレベルのチェックポイントメタデータ管理
7. OffsetSeq：バッチ内の全ソースオフセットをまとめた構造体

**関連システム・外部連携**：HDFS/S3等の分散ストレージ、StateStore（ステート管理との連携）

**権限による制御**：チェックポイントディレクトリへの読み書きアクセス権限が必要。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 17 | Streaming Query Statistics（ストリーミングクエリ統計） | 補助機能 | チェックポイント関連のメトリクス（Offset commit duration等）を表示 |

## 機能種別

データストレージ / 障害復旧

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| checkpointLocation | String | Yes | チェックポイントディレクトリパス | 書き込み可能パス |
| batchId | Long | Yes | バッチID | 0以上の連番 |
| offsetSeq | OffsetSeq | Yes | バッチのソースオフセット群 | 有効なOffsetSeq |
| commitMetadata | CommitMetadata | No | コミットメタデータ | - |

### 入力データソース

- ストリーミングエンジン（MicroBatchExecution）からのオフセット情報
- バッチ完了通知

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| OffsetSeqLog ファイル | JSON | バッチオフセット情報（offsets/ディレクトリ内） |
| CommitLog ファイル | JSON | バッチコミット情報（commits/ディレクトリ内） |
| Metadata ファイル | JSON | クエリメタデータ |

### 出力先

- チェックポイントディレクトリ（HDFS/S3/ローカルFS）
  - `{checkpointLocation}/offsets/` - OffsetSeqLog
  - `{checkpointLocation}/commits/` - CommitLog
  - `{checkpointLocation}/metadata` - クエリメタデータ
  - `{checkpointLocation}/state/` - StateStoreチェックポイント

## 処理フロー

### 処理シーケンス

```
1. チェックポイント初期化
   └─ HDFSMetadataLog でチェックポイントディレクトリを初期化
2. オフセット記録
   └─ OffsetSeqLog.add(batchId, offsetSeq) でバッチオフセットを記録
3. バッチ実行
   └─ MicroBatchExecution がバッチを実行
4. コミット記録
   └─ CommitLog.add(batchId, commitMetadata) でコミット完了を記録
5. 障害復旧時の再開
   └─ OffsetSeqLog とCommitLog を比較し、未コミットバッチから再開
6. ログ圧縮
   └─ CompactibleFileStreamLog で古いログエントリを圧縮
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[OffsetSeqLog.add - オフセット記録]
    B --> C[バッチ実行]
    C --> D{バッチ成功?}
    D -->|Yes| E[CommitLog.add - コミット記録]
    D -->|No| F[タスクリトライ]
    E --> G[次のバッチへ]
    F --> C

    H[障害復旧] --> I[OffsetSeqLog読み込み]
    I --> J[CommitLog読み込み]
    J --> K{未コミットバッチあり?}
    K -->|Yes| L[未コミットバッチから再開]
    K -->|No| M[最新コミット後から再開]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | Write-Ahead Log | オフセットはバッチ実行前に記録し、コミットはバッチ完了後に記録する（WALパターン） | 全バッチ |
| BR-02 | 障害復旧ポイント | 復旧時、OffsetSeqLogにあるがCommitLogにないバッチIDから再実行 | クエリ再起動時 |
| BR-03 | アトミック書き込み | CheckpointFileManagerがファイルのアトミックな作成を保証（create→rename） | 全ログ書き込み |
| BR-04 | ログ圧縮 | CompactibleFileStreamLogが古いログファイルを圧縮してストレージ使用量を制限 | 定期的に実行 |

### 計算ロジック

- **復旧ポイント計算**: lastCommittedBatchId = CommitLog.getLatest()のバッチID、再開バッチID = lastCommittedBatchId + 1
- **OffsetSeq構成**: Array[Option[Offset]]（各ソースのオフセット）+ MetadataのJSON形式

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| オフセット記録 | OffsetSeqLog | WRITE | バッチオフセットの永続化 |
| コミット記録 | CommitLog | WRITE | バッチコミットの永続化 |
| 読み込み | OffsetSeqLog/CommitLog | READ | 障害復旧時のログ読み込み |

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

#### OffsetSeqLog（ファイルベース）

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| WRITE | offsets/{batchId} | OffsetSeq JSON | 1バッチ1ファイル |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| CHECKPOINT_WRITE_FAILED | IOException | チェックポイントファイル書き込み失敗 | ストレージ状態を確認 |
| CORRUPT_CHECKPOINT | SparkException | チェックポイントファイルの破損 | バックアップから復旧 |
| VERSION_MISMATCH | SparkException | チェックポイントバージョンの不整合 | 互換バージョンを確認 |

### リトライ仕様

CheckpointFileManagerがファイル書き込みのアトミック性を保証。書き込み失敗時はバッチ全体がリトライされる。

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

OffsetSeqLogへの書き込みとCommitLogへの書き込みは個別にアトミック。両者の組み合わせでバッチのExactly-Once処理を保証（WALパターン）。

## パフォーマンス要件

- チェックポイント書き込みのレイテンシがバッチ処理時間に加算される
- AsyncOffsetSeqLog/AsyncCommitLogにより非同期書き込みで遅延を削減
- ログ圧縮によりチェックポイントディレクトリのファイル数を制限

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

- チェックポイントディレクトリのアクセス権限管理が必要
- チェックポイントデータにクエリのメタデータ（設定値等）が含まれる可能性

## 備考

- チェックポイント管理は `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/` 配下に実装
- ChecksumCheckpointFileManagerはチェックサム付きのファイル管理を提供
- MetadataVersionUtilによるメタデータバージョンの互換性管理

---

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

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

### 推奨読解順序

#### Step 1: データ構造を理解する

チェックポイントのデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | OffsetSeq.scala | `sql/core/.../streaming/checkpointing/` | OffsetSeq構造体（全ソースオフセット + メタデータ） |
| 1-2 | MetadataLog.scala | `sql/core/.../streaming/checkpointing/` | MetadataLogトレイト（add/get/purge等のインターフェース） |

#### Step 2: メタデータログ基盤を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | HDFSMetadataLog.scala | `sql/core/.../streaming/checkpointing/` | ファイルシステムベースのメタデータログ実装 |
| 2-2 | CheckpointFileManager.scala | `sql/core/.../streaming/checkpointing/` | アトミックファイル操作マネージャ |

#### Step 3: オフセットログとコミットログを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | OffsetSeqLog.scala | `sql/core/.../streaming/checkpointing/` | バッチオフセットログ |
| 3-2 | CommitLog.scala | `sql/core/.../streaming/checkpointing/` | バッチコミットログ |
| 3-3 | AsyncOffsetSeqLog.scala | `sql/core/.../streaming/checkpointing/` | 非同期オフセットログ |
| 3-4 | AsyncCommitLog.scala | `sql/core/.../streaming/checkpointing/` | 非同期コミットログ |

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

```
MicroBatchExecution
    │
    ├─ OffsetSeqLog.add(batchId, offsetSeq)
    │      └─ HDFSMetadataLog.add()
    │             └─ CheckpointFileManager.createAtomic()
    │
    ├─ [バッチ実行]
    │
    └─ CommitLog.add(batchId, commitMetadata)
           └─ HDFSMetadataLog.add()
                  └─ CheckpointFileManager.createAtomic()

[障害復旧時]
MicroBatchExecution
    ├─ OffsetSeqLog.getLatest() → 最新オフセットバッチID
    ├─ CommitLog.getLatest() → 最新コミットバッチID
    └─ 再開バッチID = max(lastCommittedBatchId + 1, ...)
```

### データフロー図

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

ソースオフセット          OffsetSeqLog                     offsets/{batchId}
(各ソースのOffset) ──▶  .add(batchId, offsetSeq)     ──▶  (JSON ファイル)

バッチ完了通知           CommitLog                        commits/{batchId}
(commitMetadata)  ──▶  .add(batchId, metadata)      ──▶  (JSON ファイル)

チェックポイント          HDFSMetadataLog                  復旧ポイント
(障害復旧時)       ──▶  .getLatest()                 ──▶  (batchId + offsetSeq)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| OffsetSeqLog.scala | `sql/core/.../streaming/checkpointing/` | ソース | バッチオフセットログ |
| CommitLog.scala | `sql/core/.../streaming/checkpointing/` | ソース | バッチコミットログ |
| OffsetSeq.scala | `sql/core/.../streaming/checkpointing/` | ソース | オフセットシーケンス構造体 |
| HDFSMetadataLog.scala | `sql/core/.../streaming/checkpointing/` | ソース | ファイルシステムベースログ基盤 |
| MetadataLog.scala | `sql/core/.../streaming/checkpointing/` | ソース | メタデータログインターフェース |
| CheckpointFileManager.scala | `sql/core/.../streaming/checkpointing/` | ソース | アトミックファイル管理 |
| ChecksumCheckpointFileManager.scala | `sql/core/.../streaming/checkpointing/` | ソース | チェックサム付きファイル管理 |
| AsyncOffsetSeqLog.scala | `sql/core/.../streaming/checkpointing/` | ソース | 非同期オフセットログ |
| AsyncCommitLog.scala | `sql/core/.../streaming/checkpointing/` | ソース | 非同期コミットログ |
| MetadataVersionUtil.scala | `sql/core/.../streaming/checkpointing/` | ソース | バージョン互換性ユーティリティ |
