[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-2',
    '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   |
                 |  알림)   |     | (검색/   |
                 +----------+     |  대시보드)|
                                  +----------+

실시간 대시보드 아키텍처
==========================

+----------+     +----------+     +-----------+     +----------+
| 웹/모바일|     | API      |     | Kinesis   |     | Managed  |
| 클라이언트| --> | Gateway  | --> | Data      | --> | Flink    |
+----------+     +----------+     | Streams   |     | (집계/   |
                                  +-----------+     |  분석)   |
                                                    +----------+
                                                         |
                                          +---------+----+----+---------+
                                          |         |         |         |
                                          v         v         v         v
                                     +--------+ +--------+ +--------+ +--------+
                                     |DynamoDB| |Timestream| | S3    | |CloudWatch|
                                     |(실시간 | |(시계열  | |(장기  | |(메트릭  |
                                     | 데이터)| | 데이터) | | 저장) | | 알람)   |
                                     +--------+ +--------+ +--------+ +--------+
                                          |         |
                                          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-2')
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-2')

# 컨슈머 등록
response = kinesis.register_stream_consumer(
    StreamARN='arn:aws:kinesis:ap-northeast-2: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-2: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 분할 권장