[AWS] Kinesis完全ガイド：リアルタイムストリーミングデータ処理の全て

バッチ処理 (Batch Processing)
============================
[データ収集] --> [ストレージ] --> [定期処理] --> [結果]
     |                               |
     +---- 数分 〜 数時間 -----------+

ストリーム処理 (Stream Processing)
==================================
[データ生成] --> [ストリーム] --> [即時処理] --> [結果]
     |                              |
     +---- ミリ秒 〜 数秒 ---------+

+------------------------------------------------------------------+
|                    AWS Kinesisファミリー                            |
+------------------------------------------------------------------+
|                                                                    |
|  +------------------+  +------------------+  +------------------+  |
|  | Kinesis Data     |  | Amazon Data      |  | Managed Service  |  |
|  | Streams          |  | Firehose         |  | for Apache Flink |  |
|  |                  |  |                  |  |                  |  |
|  | リアルタイム     |  | 配信パイプ      |  | ストリーム分析   |  |
|  | データ           |  | ライン           |  | (SQL, Java,      |  |
|  | ストリーミング   |  | (S3, Redshift   |  |  Python, Scala)  |  |
|  |                  |  |  OpenSearch等)   |  |                  |  |
|  +------------------+  +------------------+  +------------------+  |
|                                                                    |
|  +------------------+                                              |
|  | Kinesis Video    |                                              |
|  | Streams          |                                              |
|  |                  |                                              |
|  | ビデオ           |                                              |
|  | ストリーミング   |                                              |
|  | 取込・分析       |                                              |
|  +------------------+                                              |
+------------------------------------------------------------------+

                        Kinesis Data Stream
+----------------------------------------------------------------------+
|                                                                      |
|  Producer        Shard 1          Shard 2          Shard 3           |
|  --------     +----------+     +----------+     +----------+         |
|  |App A | --> |Record 1  | --> |Record 4  | --> |Record 7  |         |
|  |App B | --> |Record 2  | --> |Record 5  | --> |Record 8  |         |
|  |App C | --> |Record 3  | --> |Record 6  | --> |Record 9  |         |
|  --------     +----------+     +----------+     +----------+         |
|                    |                |                |                |
|                    v                v                v                |
|               Consumer A       Consumer A       Consumer A           |
|               Consumer B       Consumer B       Consumer B           |
+----------------------------------------------------------------------+

パーティションキー ハッシュ過程
================================

Partition Key: "user-123"
         |
         v
    MD5("user-123") = 0x7A3B...
         |
         v
    Hash Range: 0 ~ 2^128 - 1
         |
         v
    Shard 1: [0 ~ 2^127]          <-- この範囲にマッピング
    Shard 2: [2^127 ~ 2^128 - 1]

import boto3
import json

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

# 単一レコード送信
response = kinesis.put_record(
    StreamName='my-data-stream',
    Data=json.dumps({
        'event_type': 'page_view',
        'user_id': 'user-123',
        'page': '/products/laptop',
        'timestamp': '2026-03-20T10:30:00Z'
    }).encode('utf-8'),
    PartitionKey='user-123'
)
print(f"Shard ID: {response['ShardId']}")
print(f"Sequence Number: {response['SequenceNumber']}")

# バッチレコード送信
records = []
for i in range(100):
    records.append({
        'Data': json.dumps({
            'event_type': 'click',
            'user_id': f'user-{i % 10}',
            'element': 'buy_button',
            'timestamp': '2026-03-20T10:30:00Z'
        }).encode('utf-8'),
        'PartitionKey': f'user-{i % 10}'
    })

response = kinesis.put_records(
    StreamName='my-data-stream',
    Records=records
)
print(f"Failed records: {response['FailedRecordCount']}")

KPL 集約プロセス
=================

User Record A (100 bytes) --+
User Record B (200 bytes) --+--> Kinesis Record (500 bytes)
User Record C (200 bytes) --+         |
                                      v
User Record D (300 bytes) --+    PutRecords API
User Record E (400 bytes) --+--> (単一呼び出し)
User Record F (100 bytes) --+

import boto3
import json
import time

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

# シャードイテレーター取得
response = kinesis.get_shard_iterator(
    StreamName='my-data-stream',
    ShardId='shardId-000000000000',
    ShardIteratorType='LATEST'  # TRIM_HORIZON, AT_SEQUENCE_NUMBER, AT_TIMESTAMP等
)
shard_iterator = response['ShardIterator']

# レコードポーリング
while True:
    response = kinesis.get_records(
        ShardIterator=shard_iterator,
        Limit=100
    )

    for record in response['Records']:
        data = json.loads(record['Data'].decode('utf-8'))
        print(f"Partition Key: {record['PartitionKey']}")
        print(f"Sequence Number: {record['SequenceNumber']}")
        print(f"Data: {data}")

    shard_iterator = response['NextShardIterator']

    # GetRecordsはシャードあたり毎秒5回まで呼び出し可能
    time.sleep(0.2)

共有ファンアウト vs 拡張ファンアウト
=======================================

共有ファンアウト (Shared Fan-Out):
+--------+     +--------+
| Shard  | --> | 2MB/s  | --> Consumer A (ポーリング)
|        |     | (共有) | --> Consumer B (ポーリング)
+--------+     +--------+     Consumer C (ポーリング)

3つのコンシューマーが2MB/sを共有
各コンシューマー: ~0.67 MB/s

拡張ファンアウト (Enhanced Fan-Out):
+--------+     +--------+
| Shard  | --> | 2MB/s  | --> Consumer A (HTTP/2 プッシュ)
|        |     | 2MB/s  | --> Consumer B (HTTP/2 プッシュ)
|        |     | 2MB/s  | --> Consumer C (HTTP/2 プッシュ)
+--------+     +--------+

各コンシューマーに専用の2MB/sを割当
レイテンシ: ~70ms (共有: ~200ms+)

KCL アーキテクチャ
===================

                DynamoDB (Lease Table)
                +---------------------+
                | Shard ID | Worker   |
                |----------|----------|
                | shard-0  | worker-1 |
                | shard-1  | worker-2 |
                | shard-2  | worker-1 |
                | shard-3  | worker-2 |
                +---------------------+
                      ^         ^
                      |         |
              +-------+         +-------+
              |                         |
        +-----------+           +-----------+
        | Worker 1  |           | Worker 2  |
        | (EC2/ECS) |           | (EC2/ECS) |
        |           |           |           |
        | shard-0   |           | shard-1   |
        | shard-2   |           | shard-3   |
        +-----------+           +-----------+

シャード分割 (Split)
=====================
Shard 1 [0 ~ 100]
         |
    split at 50
         |
    +----+----+
    |         |
Shard 3    Shard 4
[0 ~ 50]  [51 ~ 100]

シャードマージ (Merge)
=======================
Shard 3 [0 ~ 50]     --> Shard 5
Shard 4 [51 ~ 100]   --> [0 ~ 100]

Amazon Data Firehose アーキテクチャ
======================================

+----------+     +-----------+     +------------+     +----------+
| ソース   |     | Firehose  |     | 変換       |     | 配信先   |
|          | --> | 配信      | --> | (任意)     | --> |          |
| - Direct |     | ストリーム|     | - Lambda   |     | - S3     |
| - Kinesis|     |           |     | - フォーマ |     | - Redshift|
|   Data   |     | バッファ: |     |   ット変換 |     | - Open-  |
|   Streams|     | - サイズ  |     | - 圧縮     |     |   Search |
| - MSK    |     | - 時間    |     | - 暗号化   |     | - Splunk |
|          |     |           |     |            |     | - HTTP   |
+----------+     +-----------+     +------------+     +----------+
                                                           |
                                                      +----------+
                                                      | バックアップ|
                                                      | S3 (原本) |
                                                      +----------+

import boto3
import json

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

# 単一レコード送信
response = firehose.put_record(
    DeliveryStreamName='my-firehose-stream',
    Record={
        'Data': json.dumps({
            'event_type': 'purchase',
            'user_id': 'user-456',
            'product': 'laptop',
            'amount': 1299.99,
            'timestamp': '2026-03-20T10:30:00Z'
        }).encode('utf-8')
    }
)

# バッチ送信
records = []
for i in range(50):
    records.append({
        'Data': json.dumps({
            'event_type': 'page_view',
            'user_id': f'user-{i}',
            'page': f'/product/{i}',
            'timestamp': '2026-03-20T10:30:00Z'
        }).encode('utf-8')
    })

response = firehose.put_record_batch(
    DeliveryStreamName='my-firehose-stream',
    Records=records
)
print(f"Failed records: {response['FailedPutCount']}")

Kinesis Video Streams アーキテクチャ
=======================================

+----------+     +------------------+     +------------------+
| デバイス |     | Kinesis Video    |     | コンシューマー   |
|          | --> | Streams          | --> |                  |
| - カメラ |     |                  |     | - HLS再生        |
| - ドローン|    | - 自動スケール   |     | - DASH再生       |
| - LIDAR  |     | - 耐久性ストレージ|    | - GetMedia API   |
| - スマート|    | - 暗号化         |     | - Rekognition    |
|   フォン |     |                  |     | - SageMaker      |
+----------+     +------------------+     +------------------+

リアルタイム分析パイプラインアーキテクチャ
=============================================

+----------+   +---------+   +----------+   +---------+   +----------+
| Web/     |   | Kinesis |   | Lambda   |   | Firehose|   | S3       |
| Mobile   |-->| Data    |-->| (変換/   |-->| 配信    |-->| (Data    |
| App      |   | Streams |   |  フィルタ)|  | ストリーム|  |  Lake)   |
+----------+   +---------+   +----------+   +---------+   +----------+
                    |                                           |
                    v                                           v
              +-----------+                              +----------+
              | Managed   |                              | Athena   |
              | Flink     |                              | (Ad-hoc  |
              | (リアル   |                              |  クエリ) |
              |  タイム分析)|                             +----------+
              +-----------+                                    |
                    |                                          v
                    v                                    +----------+
              +-----------+                              | Quick-   |
              | DynamoDB  |                              | Sight    |
              | (リアル   |                              | (BI      |
              |  タイムDB)|                              |ダッシュ  |
              +-----------+                              | ボード)  |
                                                         +----------+

# === producer.py ===
import boto3
import json
import time
import random
from datetime import datetime

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

def generate_click_event():
    """クリックストリームイベント生成"""
    pages = ['/home', '/products', '/cart', '/checkout', '/profile']
    actions = ['view', 'click', 'scroll', 'submit']
    user_id = f'user-{random.randint(1, 1000)}'

    return {
        'user_id': user_id,
        'page': random.choice(pages),
        'action': random.choice(actions),
        'session_id': f'sess-{random.randint(1, 100)}',
        'timestamp': datetime.utcnow().isoformat() + 'Z',
        'device': random.choice(['mobile', 'desktop', 'tablet']),
        'country': random.choice(['KR', 'US', 'JP', 'DE'])
    }

def send_events(batch_size=50, interval=1.0):
    """イベントをKinesisストリームに送信"""
    while True:
        records = []
        for _ in range(batch_size):
            event = generate_click_event()
            records.append({
                'Data': json.dumps(event).encode('utf-8'),
                'PartitionKey': event['user_id']
            })

        try:
            response = kinesis.put_records(
                StreamName=STREAM_NAME,
                Records=records
            )

            failed = response['FailedRecordCount']
            if failed > 0:
                print(f"Warning: {failed} records failed")

                for i, record_response in enumerate(response['Records']):
                    if 'ErrorCode' in record_response:
                        print(f"  Error: {record_response['ErrorCode']}")
            else:
                print(f"Sent {batch_size} records successfully")

        except Exception as e:
            print(f"Error: {e}")

        time.sleep(interval)

if __name__ == '__main__':
    send_events()

# === consumer.py ===
import boto3
import json
import time

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

def get_shard_ids():
    """ストリームの全シャードIDを返す"""
    response = kinesis.describe_stream(StreamName=STREAM_NAME)
    return [
        shard['ShardId']
        for shard in response['StreamDescription']['Shards']
    ]

def process_records(records):
    """レコード処理ロジック"""
    page_views = {}
    for record in records:
        data = json.loads(record['Data'].decode('utf-8'))
        page = data.get('page', 'unknown')
        page_views[page] = page_views.get(page, 0) + 1

        # 異常行動検知の例
        if data.get('action') == 'submit' and data.get('page') == '/checkout':
            print(f"[ALERT] Checkout event: user={data['user_id']}")

    for page, count in page_views.items():
        print(f"  Page: {page}, Views: {count}")

def consume_stream():
    """ストリームからデータを読み取り処理"""
    shard_ids = get_shard_ids()
    print(f"Found {len(shard_ids)} shards")

    shard_iterators = {}
    for shard_id in shard_ids:
        response = kinesis.get_shard_iterator(
            StreamName=STREAM_NAME,
            ShardId=shard_id,
            ShardIteratorType='LATEST'
        )
        shard_iterators[shard_id] = response['ShardIterator']

    while True:
        for shard_id in shard_ids:
            try:
                response = kinesis.get_records(
                    ShardIterator=shard_iterators[shard_id],
                    Limit=100
                )

                if response['Records']:
                    print(f"\n--- Shard: {shard_id} ---")
                    print(f"Records received: {len(response['Records'])}")
                    process_records(response['Records'])

                shard_iterators[shard_id] = response['NextShardIterator']

            except kinesis.exceptions.ExpiredIteratorException:
                response = kinesis.get_shard_iterator(
                    StreamName=STREAM_NAME,
                    ShardId=shard_id,
                    ShardIteratorType='LATEST'
                )
                shard_iterators[shard_id] = response['ShardIterator']

        time.sleep(1)

if __name__ == '__main__':
    consume_stream()