# 機能設計書 88-Kinesisコネクタ

## 概要

本ドキュメントは、Amazon Kinesisとの連携を提供するSparkコネクタの機能設計を記述する。Kinesisストリームからのデータ取り込みを実現する。

### 本機能の処理概要

Kinesisコネクタは、Amazon Kinesisストリームからのリアルタイムデータ取り込みをSpark Streaming（DStream API）で実現する。Kinesis Client Library（KCL）のSchedulerを使用してシャードのデータを受信し、BlockGeneratorを通じてSparkのブロックストレージに保存する。DynamoDBを利用したチェックポイント管理により、障害復旧時のデータ再取得を可能にする。

**業務上の目的・背景**：Amazon Kinesisは AWS のマネージドストリーミングサービスであり、IoTデータ、クリックストリーム、ログデータ等の大量リアルタイムデータの取り込みに使用される。SparkのKinesisコネクタにより、Kinesisストリームのデータを直接Sparkの分散処理エンジンに取り込み、リアルタイム分析やETL処理を実行できる。

**機能の利用シーン**：AWS環境でのリアルタイムデータパイプライン、IoTデータのリアルタイム分析、クリックストリーム分析、AWSサービスとSpark間のデータ連携。

**主要な処理内容**：
1. KinesisInputDStream生成：Builderパターンでストリーム設定を構築
2. KinesisReceiver起動：KCL Schedulerを起動しシャードデータを受信
3. KinesisRecordProcessor：各シャードからのデータをBlockGeneratorに挿入
4. ブロック生成：BlockGeneratorListenerでブロックの生成とシーケンス番号の記録
5. チェックポイント：DynamoDB経由でシーケンス番号のチェックポイントを管理

**関連システム・外部連携**：Amazon Kinesis Data Streams、Amazon DynamoDB（チェックポイント）、Amazon CloudWatch（メトリクス）

**権限による制御**：SparkAWSCredentialsでAWS認証情報を管理。IAMロール、アクセスキー、STS等の認証方式をサポート。

## 関連画面

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

## 機能種別

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

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| streamName | String | Yes | Kinesisストリーム名 | 非空文字列 |
| endpointUrl | String | No | KinesisエンドポイントURL | URL形式 |
| regionName | String | No | AWSリージョン名 | 有効なリージョン |
| checkpointAppName | String | Yes | チェックポイントアプリケーション名（DynamoDBテーブル名） | 非空文字列 |
| checkpointInterval | Duration | No | チェックポイント間隔 | - |
| initialPosition | KinesisInitialPosition | No | 初期読み取り位置（Latest/TrimHorizon/AtTimestamp） | 列挙値 |
| storageLevel | StorageLevel | No | ストレージレベル | - |
| messageHandler | KinesisClientRecord => T | No | メッセージ変換関数 | - |
| kinesisCreds | SparkAWSCredentials | No | Kinesis認証情報 | - |
| metricsLevel | MetricsLevel | No | CloudWatchメトリクスレベル | - |

### 入力データソース

Amazon Kinesisストリーム（KCL Scheduler経由）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| T | DStream[T] | messageHandlerで変換されたデータのDStream |
| SequenceNumberRanges | メタデータ | ブロック毎のシーケンス番号範囲 |

### 出力先

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

## 処理フロー

### 処理シーケンス

```
1. KinesisInputDStream生成
   └─ Builder.build()でパラメータを検証しDStreamを構築
2. KinesisReceiver起動
   └─ Receiver.onStart()でKCL Schedulerを起動
3. KCL Scheduler実行
   └─ 各シャードにKinesisRecordProcessorを割り当て
4. レコード処理
   └─ processRecords()でBlockGeneratorにデータを追加
5. ブロック生成
   └─ BlockGeneratorListenerでブロック生成時にシーケンス番号範囲を記録
6. ブロック保存
   └─ ブロックをSparkのBlockManagerに保存
7. チェックポイント
   └─ KinesisCheckpointerで定期的にDynamoDBにシーケンス番号を保存
8. RDD生成
   └─ createBlockRDD()でKinesisBackedBlockRDDを生成
```

### フローチャート

```mermaid
flowchart TD
    A[KinesisInputDStream.Builder.build] --> B[KinesisReceiver起動]
    B --> C[KCL Scheduler起動]
    C --> D[シャード割り当て]
    D --> E[KinesisRecordProcessor]
    E --> F[processRecords]
    F --> G[BlockGenerator.addData]
    G --> H{ブロック完成?}
    H -->|Yes| I[BlockManager保存]
    H -->|No| F
    I --> J[SequenceNumberRanges記録]
    J --> K{チェックポイント時刻?}
    K -->|Yes| L[DynamoDBチェックポイント]
    K -->|No| F
    L --> F
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-88-01 | KinesisBackedBlockRDD | すべてのブロックがSequenceNumberRangesを持つ場合、KinesisBackedBlockRDDを生成する | createBlockRDD()実行時 |
| BR-88-02 | 初期位置 | checkpointが存在しない場合、initialPositionから読み取りを開始する | 初回起動時 |
| BR-88-03 | チェックポイント管理 | 最新の正常保存済みシーケンス番号をDynamoDBにチェックポイントする | 定期実行 |
| BR-88-04 | 認証分離 | Kinesis、DynamoDB、CloudWatchに対して個別のAWS認証情報を設定可能 | 認証情報設定時 |

### 計算ロジック

特になし。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| チェックポイント | DynamoDB（checkpointAppName） | UPDATE | シーケンス番号の保存 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | 認証エラー | AWS認証情報が無効な場合 | IAMロールまたはアクセスキーを確認 |
| - | ストリーム不存在 | 指定ストリームが存在しない場合 | ストリーム名とリージョンを確認 |
| - | チェックポイントエラー | DynamoDBへの書き込みに失敗した場合 | DynamoDB権限を確認 |

### リトライ仕様

KCL Schedulerの内部リトライ機構により、一時的なエラーは自動リトライされる。

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

At-Least-Onceセマンティクスを保証する。チェックポイントの更新タイミングにより重複処理が発生し得る。

## パフォーマンス要件

- KCLのシャード並列処理により各シャードを独立してスループット最大化
- BlockGeneratorによるバッチ化でSparkへのデータ投入を効率化

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

- SparkAWSCredentialsで複数の認証方式（IAMロール、アクセスキー、STS）をサポート
- DynamoDB、CloudWatchに対する個別の認証情報設定が可能

## 備考

本コネクタはDStream APIベースであり、Spark Streaming（旧API）で使用する。Structured Streaming向けのKinesis連携は別途検討が必要。

---

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

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

### 推奨読解順序

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | KinesisInputDStream.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala` | **35-50行目**: KinesisInputDStreamのパラメータ定義（streamName, endpointUrl, regionName, initialPosition等） |
| 1-2 | SparkAWSCredentials.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/SparkAWSCredentials.scala` | AWS認証情報のデータ構造 |
| 1-3 | KinesisReadConfigurations.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReadConfigurations.scala` | 読み取り設定 |

**読解のコツ**: KinesisInputDStreamはReceiverInputDStreamを継承しており、Receiver方式でデータを受信する。Direct方式（Kafkaのような）ではないため、BlockManagerを介したデータ保存が行われる。

#### Step 2: Receiverの実装を理解する

KinesisReceiverの処理フローを追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KinesisReceiver.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala` | **47-80行目**: Receiverのドキュメント、KCL Scheduler起動、BlockGenerator連携 |

**主要処理フロー**:
1. **47-65行目**: KCL Schedulerの動作フロー説明（KinesisRecordProcessor -> BlockGenerator -> Block -> チェックポイント）
2. **80-85行目**: パラメータ定義（streamName, endpointUrl, regionName等）

#### Step 3: ブロック生成とRDD構築を理解する

ブロック生成とKinesisBackedBlockRDDの構築を追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | KinesisInputDStream.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala` | **52-86行目**: createBlockRDD()メソッド - SequenceNumberRangesの処理、KinesisBackedBlockRDD生成 |
| 3-2 | KinesisBackedBlockRDD.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala` | Kinesisベースのブロック再読み取りRDD |
| 3-3 | KinesisCheckpointer.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointer.scala` | チェックポイント管理 |

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

```
KinesisInputDStream.Builder.build()
    |
    +-- KinesisInputDStream(ssc, streamName, ...)
            |
            +-- getReceiver() -> KinesisReceiver
            |       |
            |       +-- KCL Scheduler起動
            |       |       +-- KinesisRecordProcessor (per shard)
            |       |               +-- processRecords()
            |       |               +-- BlockGenerator.addData()
            |       |
            |       +-- KinesisCheckpointer
            |               +-- DynamoDB checkpoint
            |
            +-- createBlockRDD() -> KinesisBackedBlockRDD
                    +-- SequenceNumberRanges
```

### データフロー図

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

Kinesis Stream ────▶ KCL Scheduler ──▶ KinesisRecordProcessor
  (シャード)              |                      |
                    KinesisReceiver       BlockGenerator
                          |                      |
                    SequenceNumberRanges    Block保存
                          |                      |
                          v                      v
DynamoDB ◀── チェックポイント    KinesisBackedBlockRDD ──▶ DStream[T]
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| KinesisInputDStream.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala` | ソース | DStream実装とBuilder |
| KinesisReceiver.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala` | ソース | Receiverとワーカー処理 |
| KinesisRecordProcessor.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisRecordProcessor.scala` | ソース | シャード別レコード処理 |
| KinesisBackedBlockRDD.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala` | ソース | Kinesisバックアップ付きRDD |
| KinesisCheckpointer.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointer.scala` | ソース | チェックポイント管理 |
| SparkAWSCredentials.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/SparkAWSCredentials.scala` | ソース | AWS認証情報管理 |
| KinesisReadConfigurations.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReadConfigurations.scala` | ソース | 読み取り設定 |
| KinesisUtilsPythonHelper.scala | `connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtilsPythonHelper.scala` | ソース | Python連携ヘルパー |
