[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()