📈 실시간 데이터 대시보드

왜 이걸 만들까? — 5분 전 데이터는 이미 늦었다

여러분이 라이브 커머스 플랫폼의 운영 엔지니어라고 상상해 보세요.

인기 인플루언서가 8시에 라이브 방송을 시작합니다. 방송 시작과 동시에 동시 접속자가 1만 명을 넘기고, 주문이 초당 수십 건씩 쏟아집니다. 그런데 갑자기 결제 성공률이 급락합니다.

5분 뒤에 알면? 이미 수백 명이 결제에 실패하고 이탈합니다. 매출 손실은 수천만 원. 10초 안에 알면? 즉시 결제 서비스 상태를 확인하고, 대체 결제 수단을 활성화할 수 있습니다.

이것이 실시간 데이터 대시보드의 가치입니다.

기존 배치 분석은 "어제 무슨 일이 있었는지" 사후 보고에 적합합니다. 하지만 "지금 무슨 일이 일어나고 있는지"를 파악하려면 실시간 스트리밍 파이프라인이 필요합니다.

이 실습에서는 AWS의 실시간 데이터 서비스를 조합하여 서버 모니터링 대시보드 파이프라인을 구축합니다:

• Kinesis Data Streams: 초당 수천 건의 메트릭 이벤트를 수집하는 실시간 스트림
• Lambda (프로듀서): 서버 CPU, 메모리, 요청 수 등 시뮬레이션 데이터를 생성
• Lambda (컨슈머): 스트림 레코드를 1분 단위로 집계 (평균, 최대, 최소)
• DynamoDB: 집계된 시계열 데이터를 저장 (TTL로 자동 삭제)
• API Gateway: 대시보드 클라이언트가 데이터를 조회하는 API

완성하면 1분마다 서버 메트릭이 자동으로 수집 → 집계 → 저장되고, API를 통해 최근 1시간의 CPU 사용률 추이를 조회할 수 있습니다.

이 프로젝트를 통해 배우게 되는 것:

• 실시간 스트리밍 처리: Kinesis Data Streams의 프로듀서/컨슈머 패턴
• 이벤트 소스 매핑: Kinesis → Lambda 자동 트리거 구성
• 시계열 데이터 설계: DynamoDB에서 파티션 키 + 정렬 키를 활용한 시계열 테이블
• TTL 활용: DynamoDB 자동 데이터 삭제로 비용 최적화
• 배치 vs 스트리밍: 두 처리 방식의 근본적 차이와 각각의 적합한 사용 사례

아키텍처 개요

데이터 파이프라인 흐름

비용 예측

Step 1: Kinesis Data Streams 생성

Kinesis Data Streams는 이 파이프라인의 중앙 데이터 고속도로입니다. 프로듀서가 보낸 레코드를 시간순으로 저장하고, 컨슈머가 순차적으로 읽어갈 수 있게 합니다. 온디맨드 모드를 선택하면 트래픽에 따라 자동으로 샤드 수가 조절됩니다.

aws kinesis create-stream \
  --stream-name realtime-metrics-stream \
  --stream-mode-details StreamMode=ON_DEMAND \
  --region ap-northeast-2

Step 2: DynamoDB 시계열 테이블 생성

시계열(Time Series) 데이터에 최적화된 DynamoDB 테이블을 설계합니다. 파티션 키로 메트릭 유형(cpu_usage, memory_usage 등)을, 정렬 키로 타임스탬프를 사용하면, 특정 메트릭의 시간 범위 쿼리가 매우 효율적입니다. TTL을 설정하면 오래된 데이터가 자동 삭제되어 비용을 절약합니다.

Step 3: 데이터 프로듀서 Lambda 작성

프로듀서 Lambda는 실제 서버에서 메트릭을 수집하는 에이전트를 시뮬레이션합니다. CPU 사용률(0100%), 메모리 사용률(4095%), 초당 요청 수(10~500)를 랜덤으로 생성하여 Kinesis에 전송합니다. EventBridge 규칙으로 1분마다 자동 실행되도록 설정합니다.

aws lambda create-function \
  --function-name metrics-producer \
  --runtime python3.12 \
  --handler lambda_function.lambda_handler \
  --role arn:aws:iam::YOUR_ACCOUNT:role/lambda-kinesis-role \
  --timeout 30 --memory-size 128 \
  --zip-file fileb://producer.zip

중간 점검: 프로듀서 → Kinesis 연결 확인

Step 4로 넘어가기 전에, 프로듀서가 Kinesis에 정상적으로 레코드를 전송하는지 반드시 확인합니다:

• Lambda 콘솔에서 metrics-producer를 수동 테스트 실행
• CloudWatch 로그에서 "Successfully put 3 records" 등의 성공 메시지 확인
• Kinesis 콘솔 → realtime-metrics-stream → 모니터링 탭에서 "수신 레코드" 그래프에 데이터 포인트가 찍히는지 확인
• 데이터가 보이지 않으면: Lambda 실행 역할에 AmazonKinesisFullAccess 정책이 있는지, 스트림 이름이 정확한지 확인

여기까지 성공했으면 컨슈머를 만들어 연결합니다.

Step 4: 컨슈머 Lambda (집계 로직) 작성

컨슈머 Lambda는 Kinesis 스트림에 새 레코드가 도착할 때 자동으로 트리거됩니다. 배치(batch) 단위로 레코드를 받아, 메트릭 유형별로 1분 단위 통계(평균, 최대, 최소, 데이터 포인트 수)를 계산하고, DynamoDB에 저장합니다.

Step 5: API Gateway 조회 엔드포인트

대시보드 클라이언트가 DynamoDB에 저장된 집계 데이터를 조회하는 API를 구성합니다. 쿼리 파라미터로 메트릭 유형과 시간 범위를 지정하면, 해당 범위의 시계열 데이터를 JSON 배열로 반환합니다.

# API 테스트 — 최근 1시간 CPU 사용률 조회
curl "https://YOUR_API_ID.execute-api.ap-northeast-2.amazonaws.com/dev/metrics?metric_type=cpu_usage&start_time=1700000000&end_time=1700003600"

Step 6: 통합 테스트

전체 파이프라인이 끊김 없이 동작하는지 단계별로 검증합니다. 프로듀서 → Kinesis → 컨슈머 → DynamoDB → API의 전체 흐름을 확인합니다.

핵심 개념 확인

트러블슈팅 가이드

완성 후 테스트 가이드

전체 실시간 파이프라인을 아래 체크리스트로 검증하세요:

1. 프로듀서 동작: 수동 실행 후 Kinesis 모니터링에서 레코드 수신 확인
2. 스트림 전달: Kinesis → 컨슈머 Lambda가 트리거되어 CloudWatch 로그에 처리 결과 기록
3. 집계 저장: DynamoDB 항목 탐색에서 metric_type, timestamp, avg, max, min 필드 확인
4. TTL 설정: expiry 필드가 현재 시각 + 86400(초)으로 설정되어 있는지 확인
5. API 조회: GET /metrics?metric_type=cpu_usage → 시계열 JSON 배열 반환
6. 자동 실행: EventBridge 활성화 후 5분 대기 → 5개 이상의 데이터 포인트 축적 확인
7. 지연 시간 측정: 프로듀서 실행 시각과 DynamoDB 저장 시각의 차이 측정 (목표: 30초 이내)
8. 비용 확인: Kinesis 콘솔에서 PUT 레코드 수, DynamoDB 쓰기 요청 수를 모니터링
9. 스트림 모니터링: Kinesis 메트릭에서 IteratorAge(컨슈머 지연)가 0에 가까운지 확인

확장 아이디어

1. 실시간 대시보드 UI: React + Chart.js로 프론트엔드를 만들고, 5초마다 API를 폴링하여 실시간 차트 렌더링
2. 임계값 알림: 컨슈머 Lambda에서 CPU 사용률이 90%를 넘으면 SNS로 경고 이메일 발송
3. Kinesis Data Firehose 연동: 원본 레코드를 S3에 자동 아카이빙하여, 나중에 Athena로 장기 분석
4. 다중 메트릭 대시보드: 네트워크 트래픽, 디스크 I/O, 에러율 등 메트릭을 추가하여 종합 모니터링
5. CloudWatch 커스텀 메트릭 연동: DynamoDB 대신 CloudWatch에 커스텀 메트릭으로 발행하여 CloudWatch 대시보드 활용

Kinesis 프로듀서 코드 핵심 구조

프로듀서 Lambda의 핵심 코드를 이해하면 실습이 수월합니다:

import json, random, time, os, boto3
 
kinesis = boto3.client('kinesis', region_name='ap-northeast-2')
STREAM_NAME = os.environ['STREAM_NAME']
 
def lambda_handler(event, context):
    timestamp = int(time.time())
    records = [
        {
            'Data': json.dumps({
                'metric_type': 'cpu_usage',
                'value': round(random.uniform(10, 95), 2),
                'timestamp': timestamp
            }),
            'PartitionKey': 'cpu_usage'
        },
        {
            'Data': json.dumps({
                'metric_type': 'memory_usage',
                'value': round(random.uniform(40, 90), 2),
                'timestamp': timestamp
            }),
            'PartitionKey': 'memory_usage'
        },
        {
            'Data': json.dumps({
                'metric_type': 'request_count',
                'value': random.randint(10, 500),
                'timestamp': timestamp
            }),
            'PartitionKey': 'request_count'
        }
    ]
    
    response = kinesis.put_records(
        StreamName=STREAM_NAME,
        Records=records
    )
    
    failed = response['FailedRecordCount']
    return {'statusCode': 200, 'sent': len(records), 'failed': failed}

학습 정리

리소스 정리