[AWS] Kinesis 完全指南：实时流式数据处理的一切

批处理 (Batch Processing)
============================
[数据收集] --> [存储库] --> [周期性处理] --> [结果]
     |                           |
     +---- 数分钟 ~ 数小时 ------+

流式处理 (Stream Processing)
==================================
[数据生成] --> [流] --> [即时处理] --> [结果]
     |                      |
     +---- 毫秒 ~ 数秒 -----+

+------------------------------------------------------------------+
|                    AWS Kinesis 家族                                |
+------------------------------------------------------------------+
|                                                                    |
|  +------------------+  +------------------+  +------------------+  |
|  | Kinesis Data     |  | Amazon Data      |  | Managed Service  |  |
|  | Streams          |  | Firehose         |  | for Apache Flink |  |
|  |                  |  |                  |  |                  |  |
|  | 实时数据         |  | 传输管道         |  | 流式分析         |  |
|  | 流式传输         |  | (S3, Redshift    |  | (SQL, Java,      |  |
|  |                  |  |  OpenSearch 等)  |  |  Python, Scala)  |  |
|  +------------------+  +------------------+  +------------------+  |
|                                                                    |
|  +------------------+                                              |
|  | 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-2')

# 发送单条记录
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-2')

# 获取分片迭代器
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-2')

# 发送单条记录
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      |
              |  仪表板)  |                              |  仪表板) |
              +-----------+                              +----------+

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

kinesis = boto3.client('kinesis', region_name='ap-northeast-2')
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-2')
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()