[AWS] Kinesis実践アーキテクチャ：Kafka比較とストリーミングパターン

Apache Kafka アーキテクチャ
============================

+----------+     +------------------------------------------+
| Producer | --> | Kafka Cluster                             |
+----------+     |                                          |
                 |  Broker 1    Broker 2    Broker 3        |
                 |  +-------+  +-------+  +-------+        |
                 |  |Topic A|  |Topic A|  |Topic A|        |
                 |  |Part 0 |  |Part 1 |  |Part 2 |        |
                 |  |       |  |       |  |       |        |
                 |  |Topic B|  |Topic B|  |Topic B|        |
                 |  |Part 0 |  |Part 1 |  |Part 2 |        |
                 |  +-------+  +-------+  +-------+        |
                 |                                          |
                 |  KRaft (Kafka 4.0+, ZooKeeper廃止)       |
                 +------------------------------------------+
                         |
                         v
                 +----------+
                 | Consumer |
                 | Group    |
                 +----------+

AWS Kinesis Data Streams アーキテクチャ
=========================================

+----------+     +------------------------------------------+
| Producer | --> | Kinesis Stream (完全マネージド)            |
+----------+     |                                          |
                 |  Shard 1     Shard 2     Shard 3         |
                 |  +-------+  +-------+  +-------+        |
                 |  |Records|  |Records|  |Records|        |
                 |  |1MB/s W|  |1MB/s W|  |1MB/s W|        |
                 |  |2MB/s R|  |2MB/s R|  |2MB/s R|        |
                 |  +-------+  +-------+  +-------+        |
                 |                                          |
                 |  AWS完全マネージド（インフラ管理不要）    |
                 +------------------------------------------+
                         |
                         v
                 +----------+
                 | Consumer |
                 | (KCL)    |
                 +----------+

スループット比較
=================

Kinesis（プロビジョニング）:
  書き込み: 1 MB/s x シャード数
  読み取り: 2 MB/s x シャード数（共有）
           2 MB/s x シャード数 x コンシューマー数（拡張ファンアウト）

  例: 100シャード
  書き込み: 100 MB/s
  読み取り: 200 MB/s（共有）または 200 MB/s x N（拡張ファンアウト）

Kafka（自己運用）:
  ブローカー性能に依存
  単一ブローカー: 数百 MB/s 可能
  クラスター: 数 GB/s 処理可能

  例: 6ブローカークラスター
  書き込み: 600+ MB/s
  読み取り: 数 GB/s

レイテンシ比較
===============
Kinesis:
  - PutRecord: 数十 ms
  - GetRecords（共有ファンアウト）: ~200 ms
  - Enhanced Fan-Out: ~70 ms

Kafka:
  - プロデューサー -> コンシューマー: ~2-10 ms（ネットワーク依存）
  - エンドツーエンド: ~10-50 ms

月間コスト推定（米国東部基準）
================================

シナリオ: 10 MB/s持続的データ取込、3つのコンシューマー

Kinesis（プロビジョニングモード）:
  - 10シャード必要（10 MB/s / 1 MB/s per shard）
  - シャードコスト: 10 x 0.015 x 720時間 = ~108 USD
  - PUTユニット: ~360 USD（約25.9B ユニット/月）
  - 拡張ファンアウト（3コンシューマー）: ~324 USD
  合計: ~792 USD/月

Kinesis（オンデマンドAdvantage）:
  - データ書き込み: ~25.9 TB x 0.032 = ~829 USD
  - データ読み取り: ~25.9 TB x 3 x 0.016 = ~1,243 USD
  合計: ~2,072 USD/月

Kafka（EC2自己運用）:
  - 3ブローカー（m5.xlarge）: 3 x 140 = ~420 USD
  - EBSストレージ（1TB x 3）: ~300 USD
  - 運用人員コスト: 別途
  合計: ~720 USD/月 + 運用コスト

Amazon MSK:
  - 3ブローカー（kafka.m5.large）: ~456 USD
  - ストレージ: ~300 USD
  合計: ~756 USD/月

使用シナリオ判定フローチャート
================================

データ処理要件分析
         |
    +----+----+
    |         |
同じデータを   メッセージを
複数のコンシ   一回だけ処理
ューマーが     すれば良いか？
読む必要が          |
あるか？         SQS
    |
リアルタイムの
順序保証が
必要か？
    |
    +----+----+
    |         |
   YES        NO
    |         |
 Kinesis    SQS FIFO
 Data       または
 Streams    Kinesis

Managed Flink アーキテクチャ
==============================

+-----------+     +----------------------------+     +-----------+
| ソース    |     | Managed Flink              |     | シンク    |
|           | --> |                            | --> |           |
| - Kinesis |     | +------------------------+ |     | - Kinesis |
| - MSK     |     | | Flink Application      | |     | - S3      |
| - S3      |     | |                        | |     | - DynamoDB|
|           |     | | - SQLクエリ            | |     | - Open-   |
|           |     | | - Java/Scalaアプリ     | |     |   Search  |
|           |     | | - Python (PyFlink)     | |     | - Redshift|
|           |     | |                        | |     |           |
|           |     | | ウィンドウ集約         | |     |           |
|           |     | | パターン検出           | |     |           |
|           |     | | CEP (複合イベント処理) | |     |           |
|           |     | +------------------------+ |     |           |
+-----------+     +----------------------------+     +-----------+

ウィンドウタイプ
=================

1) タンブリングウィンドウ (Tumbling Window)
   - 固定サイズ、重複なし
   |-------|-------|-------|-------|
   0       5       10      15      20 (秒)

2) スライディングウィンドウ (Sliding/Hopping Window)
   - 固定サイズ、一定間隔でスライド
   |-----------|
       |-----------|
           |-----------|
   0   2   4   6   8   10 (秒)
   サイズ: 6秒、スライド: 2秒

3) セッションウィンドウ (Session Window)
   - アクティビティベース、非アクティブ期間で区分
   |---event-event---| gap |--event-event-event--| gap |
   <-- Session 1 -->       <------ Session 2 ---->

4) グローバルウィンドウ (Global Window)
   - ストリーム全体に対する単一ウィンドウ

-- Kinesisソーステーブル定義
CREATE TABLE clickstream (
    user_id VARCHAR,
    page VARCHAR,
    action VARCHAR,
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kinesis',
    'stream' = 'clickstream-data',
    'aws.region' = 'ap-northeast-1',
    'scan.stream.initpos' = 'LATEST',
    'format' = 'json'
);

-- 1分タンブリングウィンドウでページ別閲覧数を集約
SELECT
    page,
    COUNT(*) AS view_count,
    COUNT(DISTINCT user_id) AS unique_users,
    TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,
    TUMBLE_END(event_time, INTERVAL '1' MINUTE) AS window_end
FROM clickstream
WHERE action = 'view'
GROUP BY
    page,
    TUMBLE(event_time, INTERVAL '1' MINUTE);

-- 異常行動検知: 1分以内に同一ユーザーが10回以上クリック
SELECT
    user_id,
    COUNT(*) AS click_count,
    TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start
FROM clickstream
WHERE action = 'click'
GROUP BY
    user_id,
    TUMBLE(event_time, INTERVAL '1' MINUTE)
HAVING COUNT(*) >= 10;

ログ集約アーキテクチャ
========================

+----------+     +----------+     +---------+     +----------+
| App      |     | Kinesis  |     | Firehose|     | S3       |
| Server 1 | --> | Agent    | --> |         | --> | (Raw     |
+----------+     +----------+     |         |     |  Logs)   |
                                  |         |     +----------+
+----------+     +----------+     |         |         |
| App      |     | Kinesis  |     |         |         v
| Server 2 | --> | Agent    | --> |         |     +----------+
+----------+     +----------+     |         |     | Athena   |
                                  |         |     | (Query)  |
+----------+     +----------+     |         |     +----------+
| App      |     | Kinesis  |     |         |
| Server N | --> | Agent    | --> |         |
+----------+     +----------+     +---------+
                      |
                      v
                 +----------+     +----------+
                 | Lambda   | --> | Open-    |
                 | (リアル  |     | Search   |
                 |  タイム  |     | (検索/   |
                 |  アラート)|    | ダッシュ |
                 +----------+     | ボード)  |
                                  +----------+

リアルタイムダッシュボードアーキテクチャ
==========================================

+----------+     +----------+     +-----------+     +----------+
| Web/     |     | API      |     | Kinesis   |     | Managed  |
| Mobile   | --> | Gateway  | --> | Data      | --> | Flink    |
| Client   |     |          |     | Streams   |     | (集約/   |
+----------+     +----------+     +-----------+     |  分析)   |
                                                    +----------+
                                                         |
                                          +---------+----+----+---------+
                                          |         |         |         |
                                          v         v         v         v
                                     +--------+ +--------+ +--------+ +--------+
                                     |DynamoDB| |Time-   | | S3     | |Cloud-  |
                                     |(リアル | |stream  | |(長期   | |Watch   |
                                     | タイム)| |(時系列)| | 保存)  | |(メト   |
                                     |        | |        | |        | |リック) |
                                     +--------+ +--------+ +--------+ +--------+
                                          |         |
                                          v         v
                                     +--------------------+
                                     | ダッシュボードアプリ |
                                     | (React/Vue +       |
                                     |  WebSocket)        |
                                     +--------------------+

# IoTデバイスシミュレーター
import boto3
import json
import time
import random
from datetime import datetime

kinesis = boto3.client('kinesis', region_name='ap-northeast-1')
STREAM_NAME = 'iot-sensor-data'

def simulate_sensor(device_id):
    """IoTセンサーデータシミュレーション"""
    return {
        'device_id': device_id,
        'temperature': round(random.uniform(15.0, 45.0), 2),
        'humidity': round(random.uniform(20.0, 90.0), 2),
        'pressure': round(random.uniform(990.0, 1030.0), 2),
        'battery_level': round(random.uniform(0.0, 100.0), 1),
        'location': {
            'lat': round(random.uniform(33.0, 38.0), 6),
            'lon': round(random.uniform(126.0, 130.0), 6)
        },
        'timestamp': datetime.utcnow().isoformat() + 'Z'
    }

def ingest_iot_data(num_devices=100, interval=1.0):
    """IoTデータをKinesisに取込"""
    device_ids = [f'sensor-{i:04d}' for i in range(num_devices)]

    while True:
        records = []
        for device_id in device_ids:
            sensor_data = simulate_sensor(device_id)
            records.append({
                'Data': json.dumps(sensor_data).encode('utf-8'),
                'PartitionKey': device_id
            })

        # PutRecordsは最大500件まで
        for batch_start in range(0, len(records), 500):
            batch = records[batch_start:batch_start + 500]
            response = kinesis.put_records(
                StreamName=STREAM_NAME,
                Records=batch
            )

            failed = response['FailedRecordCount']
            if failed > 0:
                print(f"Batch failed: {failed} records")
                # 指数バックオフリトライロジック
                retry_records = []
                for i, result in enumerate(response['Records']):
                    if 'ErrorCode' in result:
                        retry_records.append(batch[i])
                if retry_records:
                    time.sleep(0.5)
                    kinesis.put_records(
                        StreamName=STREAM_NAME,
                        Records=retry_records
                    )

        print(f"Ingested {len(records)} sensor readings")
        time.sleep(interval)

if __name__ == '__main__':
    ingest_iot_data()

イベントソーシングアーキテクチャ
==================================

+----------+     +----------+     +-----------+
| Command  |     | Kinesis  |     | Event     |
| Handler  | --> | Data     | --> | Processor |
|          |     | Streams  |     | (Lambda/  |
| - 注文   |     | (イベント|     |  ECS)     |
| - 決済   |     |  ストア) |     +-----------+
| - 配送   |     +----------+          |
+----------+          |          +-----+-----+
                      |          |           |
                      v          v           v
                 +----------+ +--------+ +--------+
                 | S3       | |DynamoDB| |SNS     |
                 | (イベント| |(読み取り| |(通知)  |
                 |  アーカイブ)| | モデル)| |        |
                 +----------+ +--------+ +--------+

イベントフロー例:
1. OrderCreated -> 注文作成イベント
2. PaymentProcessed -> 決済処理イベント
3. InventoryReserved -> 在庫予約イベント
4. ShipmentCreated -> 配送作成イベント

MLフィーチャーパイプライン
============================

+----------+     +----------+     +-----------+     +----------+
| イベント |     | Kinesis  |     | Managed   |     | Feature  |
| ソース   | --> | Data     | --> | Flink     | --> | Store    |
|          |     | Streams  |     | (フィー   |     | (Sage-   |
| - クリック|    |          |     |  チャー計算)|    |  Maker)  |
| - 購入   |     |          |     |           |     +----------+
| - 検索   |     |          |     | リアルタイム:|       |
+----------+     +----------+     | - セッション数|      v
                                  | - 最近の   |     +----------+
                                  |   購入数   |     | MLモデル |
                                  | - 平均     |     | 推論     |
                                  |   滞在時間 |     +----------+
                                  +-----------+

良いパーティションキー例
========================

1) ユーザーID（高カーディナリティ）
   user-001 -> Shard 1
   user-002 -> Shard 3
   user-003 -> Shard 2
   ...
   均等に分散

2) UUID（最高分散）
   ランダムUUID -> 完璧な分散
   欠点: 同じエンティティの順序保証不可

3) 複合キー
   "region-userType-userId"
   きめ細かい分散制御が可能

悪いパーティションキー例
========================

1) 日付 ("2026-03-20")
   全レコードが同じシャードへ -> ホットシャード

2) 国コード ("KR", "US", "JP")
   カーディナリティが低すぎる
   特定の国にトラフィックが偏る

3) 固定値 ("default")
   単一シャードに全負荷が集中

KPL集約最適化
==============

集約なし:
Record 1 (100B) -> PutRecord -> 1 API呼び出し
Record 2 (200B) -> PutRecord -> 1 API呼び出し
Record 3 (150B) -> PutRecord -> 1 API呼び出し
合計: 3 API呼び出し、450B送信

KPL集約使用:
Record 1 (100B) --+
Record 2 (200B) --+--> 集約レコード (450B) -> 1 API呼び出し
Record 3 (150B) --+
合計: 1 API呼び出し、450B送信

効果:
- API呼び出し数の大幅削減
- PUTユニットコスト削減
- スループットの大幅向上

# 拡張ファンアウトコンシューマー登録
import boto3

kinesis = boto3.client('kinesis', region_name='ap-northeast-1')

# コンシューマー登録
response = kinesis.register_stream_consumer(
    StreamARN='arn:aws:kinesis:ap-northeast-1:123456789012:stream/my-stream',
    ConsumerName='analytics-consumer'
)
consumer_arn = response['Consumer']['ConsumerARN']
print(f"Consumer ARN: {consumer_arn}")

# コンシューマー状態確認
response = kinesis.describe_stream_consumer(
    StreamARN='arn:aws:kinesis:ap-northeast-1:123456789012:stream/my-stream',
    ConsumerName='analytics-consumer'
)
print(f"Status: {response['ConsumerDescription']['ConsumerStatus']}")

import time
import random

def put_records_with_retry(kinesis_client, stream_name, records, max_retries=3):
    """指数バックオフを使用したPutRecordsリトライ"""

    for attempt in range(max_retries):
        response = kinesis_client.put_records(
            StreamName=stream_name,
            Records=records
        )

        failed_count = response['FailedRecordCount']

        if failed_count == 0:
            return response

        # 失敗したレコードのみ抽出
        retry_records = []
        for i, result in enumerate(response['Records']):
            if 'ErrorCode' in result:
                error_code = result['ErrorCode']
                if error_code == 'ProvisionedThroughputExceededException':
                    retry_records.append(records[i])
                else:
                    print(f"Non-retryable error: {error_code}")

        if not retry_records:
            return response

        records = retry_records

        # 指数バックオフ + ジッター
        backoff = min(2 ** attempt * 0.1, 5.0)
        jitter = random.uniform(0, backoff * 0.5)
        wait_time = backoff + jitter
        print(f"Retry {attempt + 1}: {len(retry_records)} records, waiting {wait_time:.2f}s")
        time.sleep(wait_time)

    print(f"Failed after {max_retries} retries: {len(records)} records")
    return None

拡張モニタリング有効化時の追加メトリクス
==========================================

シャードレベルメトリクス:
- IncomingBytes（シャード別）
- IncomingRecords（シャード別）
- IteratorAgeMilliseconds（シャード別）
- OutgoingBytes（シャード別）
- OutgoingRecords（シャード別）
- ReadProvisionedThroughputExceeded（シャード別）
- WriteProvisionedThroughputExceeded（シャード別）

ホットシャード検出:
  Shard 1: IncomingBytes = 200 KB/s  [正常]
  Shard 2: IncomingBytes = 950 KB/s  [警告! 限界付近]
  Shard 3: IncomingBytes = 300 KB/s  [正常]
  -> Shard 2の分割を推奨