# 機能設計書 37-Kafkaソース/シンク

## 概要

本ドキュメントは、Apache SparkのStructured StreamingにおけるKafka連携機能に関する機能設計書である。KafkaトピックからのストリームデータのMicroBatch/Continuous読み取り、およびKafkaトピックへのストリーム/バッチ書き込みを提供する。

### 本機能の処理概要

**業務上の目的・背景**：Apache Kafkaは分散メッセージングの標準的プラットフォームとして広く利用されている。SparkのKafkaソース/シンク連携により、Kafkaをイベントソースとするリアルタイムデータパイプライン、ETL処理、ストリーム分析をStructured Streaming APIで構築可能にする。バッチ読み書きにも対応し、Kafkaに蓄積された履歴データの分析も実現する。

**機能の利用シーン**：Kafkaトピックからのリアルタイムイベント処理、ストリーム分析結果のKafkaトピックへの書き戻し、Kafkaデータの履歴バッチ分析、イベント駆動マイクロサービスのデータ処理層。

**主要な処理内容**：
1. KafkaSourceProvider：Kafkaデータソースの登録とインスタンス生成（shortName="kafka"）
2. KafkaMicroBatchStream：マイクロバッチモードでのKafkaトピック読み取り
3. KafkaContinuousStream：連続処理モードでのKafkaトピック読み取り
4. KafkaStreamingWrite/KafkaBatchWrite：Kafkaトピックへのストリーム/バッチ書き込み
5. KafkaOffsetReader：Kafkaブローカーからのオフセット情報取得（Consumer API/Admin API）
6. KafkaDataConsumer：Kafkaコンシューマーのプーリングとデータフェッチ
7. ConsumerStrategy：サブスクリプション戦略（assign/subscribe/subscribePattern）

**関連システム・外部連携**：Apache Kafkaブローカー、Kafka Consumer/Producer API、Schema Registry（Avro/Protobuf連携時）

**権限による制御**：Kafkaブローカーの認証（SASL/SSL）、ACL設定に従う。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能に直接関連するWeb UI画面はない（ストリーミングクエリ統計画面で間接的に表示） |

## 機能種別

データ連携 / リアルタイム処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| kafka.bootstrap.servers | String | Yes | Kafkaブローカーアドレス | host:port形式 |
| subscribe / assign / subscribePattern | String | Yes（択一） | トピック指定方法 | 有効なトピック名/パターン |
| startingOffsets | String | No | 開始オフセット（earliest/latest/JSON） | デフォルト: latest（ストリーム）/earliest（バッチ） |
| endingOffsets | String | No | 終了オフセット（latest/JSON） | バッチ読み取り時のみ |
| includeHeaders | String | No | ヘッダー情報を含むか | true/false |
| failOnDataLoss | String | No | データロス時にフェイルするか | デフォルト: true |

### 入力データソース

- Apache Kafkaトピック（メッセージ：key, value, topic, partition, offset, timestamp, headers）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| key | BinaryType | メッセージキー |
| value | BinaryType | メッセージ値 |
| topic | StringType | トピック名 |
| partition | IntegerType | パーティション番号 |
| offset | LongType | オフセット |
| timestamp | TimestampType | タイムスタンプ |
| timestampType | IntegerType | タイムスタンプ種別 |
| headers | ArrayType | メッセージヘッダー（オプション） |

### 出力先

- Spark DataFrame（読み取り時）
- Kafkaトピック（書き込み時）

## 処理フロー

### 処理シーケンス

```
1. データソース初期化
   └─ KafkaSourceProvider.shortName()="kafka" で登録
2. ストリーム読み取り開始
   └─ KafkaMicroBatchStream.latestOffset() で最新オフセット取得
3. オフセット範囲計算
   └─ KafkaOffsetRangeCalculator でパーティション別読み取り範囲を決定
4. データフェッチ
   └─ KafkaDataConsumer でパーティション単位にメッセージを読み取り
5. レコード変換
   └─ KafkaRecordToRowConverter でKafkaレコードをSpark行に変換
6. ストリーム書き込み
   └─ KafkaStreamingWrite / KafkaDataWriter でメッセージをProducerで送信
7. オフセットコミット
   └─ チェックポイントにオフセットを記録
```

### フローチャート

```mermaid
flowchart TD
    A[KafkaSourceProvider] --> B{操作種別}
    B -->|ストリーム読み取り| C[KafkaMicroBatchStream]
    B -->|連続処理読み取り| D[KafkaContinuousStream]
    B -->|バッチ読み取り| E[KafkaBatch]
    B -->|ストリーム書き込み| F[KafkaStreamingWrite]
    B -->|バッチ書き込み| G[KafkaBatchWrite]
    C --> H[KafkaOffsetReader]
    H --> I[latestOffset/initialOffset]
    I --> J[KafkaOffsetRangeCalculator]
    J --> K[KafkaBatchPartitionReader]
    K --> L[KafkaDataConsumer]
    L --> M[KafkaRecordToRowConverter]
    F --> N[KafkaDataWriter]
    N --> O[CachedKafkaProducer]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | 固定スキーマ | Kafkaソースは固定スキーマ（key/value/topic/partition/offset/timestamp等）を持ち、カスタムスキーマ指定不可 | sourceSchema()時 |
| BR-02 | サブスクリプション戦略 | assign（パーティション指定）、subscribe（トピック指定）、subscribePattern（パターン指定）の3つから1つを択一 | データソース生成時 |
| BR-03 | データロス検知 | failOnDataLoss=trueの場合、オフセットの欠損を検知してエラー発生 | オフセットチェック時 |
| BR-04 | Producerキー/値制約 | 書き込み時、key/valueカラムはBinaryTypeまたはStringType | KafkaWrite時 |

### 計算ロジック

- **オフセット範囲計算**: KafkaOffsetRangeCalculatorが各パーティションの(startOffset, endOffset)を計算し、minPartitions設定に基づいてパーティション分割を最適化
- **Producerプーリング**: CachedKafkaProducerがKafkaProducerインスタンスをキャッシュし、接続コストを削減

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし（Kafkaブローカーとの通信） |

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

データベース操作は発生しない。Kafkaブローカーに対するConsumer/Producer API呼び出し。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| DATA_LOSS | IllegalStateException | failOnDataLoss=trueでオフセット欠損 | データロス許容またはトピック確認 |
| KAFKA_CONNECTION | KafkaException | ブローカー接続失敗 | ブローカー設定を確認 |
| MISSING_OPTIONS | IllegalArgumentException | 必須オプション未指定 | 必須オプションを設定 |

### リトライ仕様

Kafka Consumer/ProducerのネイティブリトライメカニズムとSparkのタスクリトライが併用される。

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

Kafkaの書き込みはProducerのackレベル（acks=all等）で信頼性を制御。ストリーミングのExactly-once保証はチェックポイント+冪等Producerで実現。

## パフォーマンス要件

- Producerプーリング（CachedKafkaProducer/InternalKafkaProducerPool）により接続オーバーヘッドを削減
- Consumerプーリング（InternalKafkaConsumerPool/FetchedDataPool）によりフェッチ効率を向上
- minPartitionsオプションでKafkaパーティション以上の並列度を設定可能

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

- SASL/SSL認証のサポート（kafka.* プレフィックス設定）
- KafkaConfigUpdaterによるセキュリティ関連設定の管理

## 備考

- Kafkaコネクタは `connector/kafka-0-10-sql/` 配下に独立モジュールとして配置
- Kafka 0.10以降のAPIに対応
- KafkaOffsetReaderにはConsumerベース（KafkaOffsetReaderConsumer）とAdminベース（KafkaOffsetReaderAdmin）の2実装がある

---

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

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

### 推奨読解順序

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

Kafkaメッセージのスキーマとオフセット管理構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | KafkaRecordToRowConverter.scala | `connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/` | kafkaSchema定義（固定スキーマ） |
| 1-2 | KafkaSourceOffset.scala | `connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/` | オフセット表現 |
| 1-3 | KafkaOffsetRangeLimit.scala | `connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/` | オフセット範囲指定 |

#### Step 2: エントリーポイントを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KafkaSourceProvider.scala | `connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/` | shortName()="kafka"、sourceSchema()、createSource() |

**主要処理フロー**:
- **52-58行目（KafkaSourceProvider.scala）**: KafkaSourceProviderが多数のプロバイダートレイトを実装
- **61行目**: shortName()="kafka"
- **67-77行目**: sourceSchema() - 固定スキーマを返却、カスタムスキーマ不可

#### Step 3: 読み取りと書き込みを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | KafkaMicroBatchStream.scala | `connector/kafka-0-10-sql/...` | マイクロバッチ読み取り |
| 3-2 | KafkaStreamingWrite.scala | `connector/kafka-0-10-sql/...` | ストリーム書き込み |
| 3-3 | KafkaDataWriter.scala | `connector/kafka-0-10-sql/...` | データライター |
| 3-4 | KafkaDataConsumer.scala | `connector/kafka-0-10-sql/.../consumer/` | データコンシューマー |

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

```
spark.readStream.format("kafka").load()
    │
    └─ KafkaSourceProvider
           ├─ KafkaMicroBatchStream / KafkaContinuousStream
           │      ├─ KafkaOffsetReader (Admin/Consumer)
           │      ├─ KafkaOffsetRangeCalculator
           │      └─ KafkaBatchPartitionReader
           │             └─ KafkaDataConsumer
           │                    └─ InternalKafkaConsumerPool
           │
           └─ KafkaStreamingWrite / KafkaBatchWrite
                  └─ KafkaDataWriter
                         └─ CachedKafkaProducer
                                └─ InternalKafkaProducerPool
```

### データフロー図

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

Kafkaトピック             KafkaOffsetReader              DataFrame
(メッセージ)       ──▶  → KafkaDataConsumer          ──▶  (key/value/topic/
                        → KafkaRecordToRowConverter        partition/offset)

DataFrame                KafkaDataWriter                  Kafkaトピック
(key/value)       ──▶  → CachedKafkaProducer        ──▶  (メッセージ)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| KafkaSourceProvider.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | データソースプロバイダー |
| KafkaMicroBatchStream.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | マイクロバッチストリーム |
| KafkaContinuousStream.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | 連続処理ストリーム |
| KafkaStreamingWrite.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | ストリーム書き込み |
| KafkaBatchWrite.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | バッチ書き込み |
| KafkaDataWriter.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | データライター |
| KafkaDataConsumer.scala | `connector/kafka-0-10-sql/.../kafka010/consumer/` | ソース | データコンシューマー |
| KafkaOffsetReader.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | オフセットリーダー基底 |
| KafkaOffsetReaderAdmin.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | Admin APIベースオフセットリーダー |
| KafkaOffsetReaderConsumer.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | Consumer APIベースオフセットリーダー |
| KafkaOffsetRangeCalculator.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | オフセット範囲計算 |
| ConsumerStrategy.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | サブスクリプション戦略 |
| CachedKafkaProducer.scala | `connector/kafka-0-10-sql/.../kafka010/producer/` | ソース | Producerキャッシュ |
| KafkaRecordToRowConverter.scala | `connector/kafka-0-10-sql/.../kafka010/` | ソース | レコード→行変換 |
