# 機能設計書 87-Kafkaコネクタ

## 概要

本ドキュメントは、Apache Kafkaとの連携を提供するSparkコネクタの機能設計を記述する。Kafkaトピックの読み書きおよびKafka認証トークンの管理を行う。

### 本機能の処理概要

Kafkaコネクタは、Apache KafkaのトピックからSpark DStreamまたはRDDとしてデータを読み取る機能を提供する（kafka-0-10モジュール）。Direct方式によりKafkaパーティションとSparkパーティションを1:1でマッピングし、Exactly-Once処理セマンティクスを実現する。

**業務上の目的・背景**：Apache Kafkaはイベントストリーミングの事実上の標準であり、リアルタイムデータパイプラインの中核として広く使用されている。SparkのKafkaコネクタは、Kafkaからのデータ取り込みとSparkの分散データ処理を組み合わせることで、リアルタイムデータ分析やETL処理を実現する。Direct方式によりレシーバーを使わずKafkaの各パーティションから直接データを読み取り、効率性と正確性を両立する。

**機能の利用シーン**：Kafkaトピックからのリアルタイムデータ取り込み、イベントストリーム処理、ログ集約と分析、Kafka-to-Sparkリアルタイムパイプライン。

**主要な処理内容**：
1. DirectKafkaInputDStream：Direct方式でKafkaパーティションからデータを読み取るDStream
2. KafkaRDD：特定のオフセット範囲からバッチ的にデータを読み取るRDD
3. オフセット管理：currentOffsetsでパーティション別の消費位置を追跡
4. コミット：CanCommitOffsetsインターフェースでオフセットのKafkaへのコミット
5. レート制御：BackPressureとmaxRatePerPartitionでデータ取り込みレートを制御

**関連システム・外部連携**：Apache Kafka（ブローカー、Consumer API）、Spark Streaming（DStream API）

**権限による制御**：Kafka認証（SASL/SSL）設定をkafkaParamsで渡す。Kafkaセキュリティトークンの管理をサポート。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能はコネクタライブラリであり直接関連する画面はない |

## 機能種別

データ連携 / ストリーム処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| kafkaParams | Map[String, Object] | Yes | Kafka Consumer設定（bootstrap.servers等） | bootstrap.servers必須 |
| consumerStrategy | ConsumerStrategy | Yes | Subscribe/SubscribePattern/Assign | - |
| locationStrategy | LocationStrategy | Yes | PreferConsistent/PreferBrokers/PreferFixed | - |
| offsetRanges | Array[OffsetRange] | バッチのみ | 読み取り範囲 | - |
| spark.streaming.kafka.maxRatePerPartition | Long | No | パーティションあたり最大レート | 正の整数 |
| spark.streaming.backpressure.initialRate | Long | No | バックプレッシャー初期レート | 正の整数 |

### 入力データソース

Kafkaトピック（Consumer API経由）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| ConsumerRecord[K, V] | DStream/RDD | Kafkaメッセージ（キー、値、トピック、パーティション、オフセット） |
| offsetRanges | Array[OffsetRange] | 処理済みオフセット範囲 |

### 出力先

Spark DStream/RDD（アプリケーション内での後続処理）

## 処理フロー

### 処理シーケンス

```
1. DStream/RDD生成
   └─ KafkaUtils.createDirectStream() or createRDD()
2. Consumer初期化
   └─ consumerStrategy.onStart()でKafka Consumerを生成
3. オフセット範囲計算
   └─ currentOffsetsとlatestOffsetsの差分を計算
4. レート制限適用
   └─ maxRatePerPartitionおよびbackpressure設定を適用
5. KafkaRDD生成
   └─ 各パーティションの(fromOffset, untilOffset)でRDDを構築
6. データ取得
   └─ KafkaDataConsumerでKafkaからデータをフェッチ
7. オフセットコミット（オプション）
   └─ commitAsync()でオフセットをKafkaにコミット
```

### フローチャート

```mermaid
flowchart TD
    A[createDirectStream] --> B[Consumer初期化]
    B --> C[オフセット範囲計算]
    C --> D{レート制限?}
    D -->|Yes| E[maxRatePerPartition適用]
    D -->|No| F[全オフセット取得]
    E --> G[KafkaRDD生成]
    F --> G
    G --> H[パーティション別データ取得]
    H --> I[ConsumerRecord DStream]
    I --> J{コミット要?}
    J -->|Yes| K[commitAsync]
    J -->|No| L[完了]
    K --> L
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-87-01 | Direct方式 | Kafkaパーティション=RDDパーティションの1:1マッピング | 常時 |
| BR-87-02 | オフセット追跡 | currentOffsetsでパーティション別の消費位置を記録する | DStream compute() |
| BR-87-03 | ブローカー情報取得 | getBrokersメソッドでパーティション-ブローカーホストマッピングを取得 | PreferBrokers使用時 |
| BR-87-04 | ConsumerRecord非シリアライズ | ConsumerRecordはシリアライズ不可。persistやwindow前にmapで値を抽出する必要がある | persist/window呼び出し時 |

### 計算ロジック

オフセット範囲計算：各パーティションについて `untilOffset = min(latestOffset, currentOffset + maxMessagesPerPartition)` でバッチあたりの処理範囲を決定する。

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

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

本機能はデータベース操作を行わない。Kafkaトピックの読み取りのみ。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | Consumer初期化エラー | Kafkaブローカーに接続できない場合 | bootstrap.servers設定を確認 |
| - | オフセット範囲外 | 指定オフセットがKafka上に存在しない場合 | auto.offset.reset設定で制御 |
| - | 認証エラー | SASL/SSL認証に失敗した場合 | 認証設定を確認 |

### リトライ仕様

KafkaDataConsumerがKafka Consumer APIの内部リトライ機構を活用する。

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

Exactly-Once処理はオフセット管理により実現する。KafkaへのオフセットコミットはAt-Least-Onceであるが、アプリケーション側でオフセットを管理することでExactly-Onceを実現可能。

## パフォーマンス要件

- Direct方式によりレシーバーボトルネックを回避
- パーティション並列性によるスループット最大化
- maxRatePerPartitionによるバックプレッシャー制御

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

- kafkaParamsでSASL/SSL認証設定を指定可能
- fixKafkaParams()でセキュリティ関連設定の自動補正

## 備考

本コネクタ（kafka-0-10）はDStream APIベースである。Structured Streaming向けのKafka連携は別モジュール（kafka-0-10-sql）で提供される。

---

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

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

### 推奨読解順序

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

Kafkaコネクタのコアデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | OffsetRange.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/OffsetRange.scala` | オフセット範囲（topic, partition, fromOffset, untilOffset）の定義 |
| 1-2 | ConsumerStrategy.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/ConsumerStrategy.scala` | Subscribe/SubscribePattern/Assignの各戦略 |
| 1-3 | LocationStrategy.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/LocationStrategy.scala` | PreferConsistent/PreferBrokers/PreferFixedの各戦略 |

**読解のコツ**: OffsetRangeがKafkaRDDの各パーティションの読み取り範囲を定義する中核データ構造。

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

KafkaUtilsのファクトリメソッドを追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KafkaUtils.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaUtils.scala` | **52-71行目**: createRDD()メソッド - バッチ読み取り、PreferBrokers禁止ルール |

**主要処理フロー**:
1. **58-65行目**: LocationStrategyの評価、preferredHostsの決定
2. **66-67行目**: kafkaParamsのコピーとfixKafkaParams()適用
3. **70行目**: KafkaRDD生成

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

ストリーミング読み取りの主要ロジックを追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | DirectKafkaInputDStream.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala` | **53-100行目**: DStream初期化、Consumer管理、オフセット追跡 |
| 3-2 | KafkaRDD.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaRDD.scala` | RDDパーティション生成とデータ読み取り |
| 3-3 | KafkaDataConsumer.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaDataConsumer.scala` | 実際のKafkaデータ取得処理 |

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

```
KafkaUtils.createDirectStream() / createRDD()
    |
    +-- DirectKafkaInputDStream
    |       +-- consumer() [Kafka Consumer初期化]
    |       +-- compute(time) [バッチ生成]
    |       |       +-- latestOffsets()
    |       |       +-- clamp() [レート制限]
    |       |       +-- new KafkaRDD()
    |       |
    |       +-- commitAsync() [オフセットコミット]
    |
    +-- KafkaRDD
            +-- KafkaRDDPartition (per Kafka partition)
            +-- KafkaDataConsumer.get()
                    +-- consumer.poll() / seek()
```

### データフロー図

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

Kafkaトピック ────▶ Consumer.poll() ──▶ KafkaRDD
  (パーティション)       |                  (パーティション)
                   KafkaDataConsumer           |
                         |                    v
                   OffsetRange ──▶ ConsumerRecord DStream
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| KafkaUtils.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaUtils.scala` | ソース | ファクトリメソッド |
| DirectKafkaInputDStream.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala` | ソース | Direct DStream実装 |
| KafkaRDD.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaRDD.scala` | ソース | Kafka RDD実装 |
| KafkaRDDPartition.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaRDDPartition.scala` | ソース | パーティション定義 |
| KafkaDataConsumer.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaDataConsumer.scala` | ソース | Kafkaデータ取得 |
| ConsumerStrategy.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/ConsumerStrategy.scala` | ソース | Consumer戦略 |
| LocationStrategy.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/LocationStrategy.scala` | ソース | ロケーション戦略 |
| OffsetRange.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/OffsetRange.scala` | ソース | オフセット範囲 |
| PerPartitionConfig.scala | `connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/PerPartitionConfig.scala` | ソース | パーティション別設定 |
