[AWS] 실시간의 실시간 - CloudFront, Amazon Kinesis Data Streams

이번 포스팅에서는 Cloudwave교육과정 최종 프로젝트로서 올리브영 시스템을 구축하는 과제를 진행하며 제안한 개선사항에 대한 내용을 다루겠습니다.

---

1. 요구사항 정의

썸네일과 같이 올리브영의 현재 시스템 중에는 실시간 랭킹이라는 기능이 있습니다. 이는 앱에서만 지원하는 기능인 것으로 확인이 됩니다(아마도..!) 해당 기능은 현재 어떤 상품이 인기가 있는 지 보여주는 역할을 합니다. 

하지만 이 기능을 보면 볼수록 한가지 의아한 부분이 있었습니다. 바로 실시간이라고 하지만 사실 그 랭킹은 몇 시간 전에 업데이트 된 랭킹일 수 있다는 점이었습니다. 저는 실시간이라는 키워드와 2시간 전 업데이트라는 키워드가 이중적이라고 생각했습니다. 그렇다면 올리브영은 왜 몇 시간 전 상품에도 실시간이라는 키워드를 넣어야만 했을까요?

출처: 넷플릭스 흑백요리사

 
제가 생각하기에 실시간이라는 키워드가 꽤나 매력적이었을 것이라 생각합니다. 그렇다고 리얼타임으로 기능을 제공한다면 사용자 경험이 분명 떨어질 것이라 생각했습니다. 극단적인 예로 마음에 담아두었던 1위 상품이 잠깐 한눈 판 사이에 순위권에서 사라져버려 찾을 수 없을 수 없을 지도 모르니까요. 이런저런 생각을 해봐도 올리브영의 약간은 애매한 실시간 기능은 꽤 합리적이라는 생각이 듭니다. 하지만 그럼에도 저는 실시간이라는 키워드가 이렇게 소비되는 것은 조금 아까웠습니다.
 

2. 제안사항 - “실시간의 실시간 / 트렌드 세터”의 이미지를 살리자!

출처: 구글 검생 n분만에 완판

 
위 스크린샷은 광고의 효과로 훌륭한 비즈니스 효과를 낸 것으로 보입니다. 각각 5분 1분으로 꽤나 짧은 시간동안 판매가 된 것을 볼 수 있습니다. 광고가 아니더라도 방송에 유명 아이돌이 특정 제품을 사용하는 장면이 미디어에 노출되면 금세 해당 상품이 품절되는 기사를 많이 접해봤을 것으로 생각합니다. 이처럼 실시간과 트렌드는 매출증가와 직결되어 있습니다. 이를 위해 실시간랭킹보다 더 실시간 랭킹을 제공하는 기능을 구현하고자 기획했습니다.
 

3.  아키텍쳐 설계

수정 전 무지성 기획
사실 가장 처음 생각했던 방법은 DB를 활용한 방법이었습니다. 그나마 효율적인 방법을 고려해보고자 RDS의 구매내역 데이터를 생성할 때 복합키(시계열 & 아이템ID)로 DynamoDB에 Count하는 방식이었습니다. 굳이 db를 나눈 이유는 조회한 데이터를 쿼리만 하는 DB를 분리하여 I/O연산과 쿼리연산을 최소할 수 있지 않을까 생각했기 때문입니다. 그렇게 전체적인 흐름은 아래 그림과 같게 구현되었습니다. 이에 따라 정상적인 요청이 들어왔을 때 5분마다 다른 랭킹 화면을 볼 수 있게 됩니다. 

DB 연산

하지만 그럼에도 대규모 트래픽을 고려하면, DB 부하와 DynamoDB 비용 등등 5분마다 업데이트 되는 데이터는 로그 활용됨이 맞다고 생각했습니다. 

수정 전 예상 비용

이에 따라 CloudFront에서 생성된 로그에 따른 데이터 처리를 고려했습니다. 간단하게 설명하자면 트래픽이 CloudFront를 통해 들어오면 Amozon Kinesis, Firehose, S3, Glue, Athena 등을 통해 분석한 결과를 5분마다 Elasticache에 업데이트 함으로써 효율적인 실시간의 실시간 시스템을 구축하기로 계획했습니다.

kinesis 아키텍처

 

4. AWS Kinesis Data Streams 구현

우리는 CloudFront의 로그를 활용하기로 했습니다. 그럼 먼저 로그를 남기는 방식에 대해서 정해야 합니다. CloudFront의 로깅 방식에는 표준 로그, 실시간 로그가 있습니다.(엣지 함수, AWS CloudTrail 도 있기는 함) 어떤 기술이든 첫 경험은 basic을 선택하는 저이기에 평소의 저였다면 표준 로깅 방식을 활용했을 것입니다. 하지만 기술에 대한 설명을 읽어보던 중 다음 대목을 발견했습니다.
 

출처: AWS Docs - Cloudfront

 
이처럼 Cloudfront의 표준 로그 방식에는 큰 단점이 있었습니다. 바로 24시간까지 지연되기도 한다는 점이었는데요. 이는 저희가 세운 실시간의 실시간이라는 타이틀을 전혀 사용할 수 없는 방법이었습니다.
 
 

출처: AWS Docs - Cloudfront

 
이에 따라 이름부터 매력적인 실시간 로그를 사용하고자 계획했습니다. 이제 사설은 그만하고 실습을 진행해보겠습니다.
 
1. CloudFront 실시간 로그 구성 설정

Cloudfront 실시간 로그 구성

 
- CloudFront에서 수집할 필드를 먼저 설정해줍니다. 사실 CloudFront의 로그 이전에 2번인 Kinesis 먼저 구성을 완료한 상태에서 로그를 연결해주어야 합니다. CloudFront에서 생성된 로그가 Kinesis에 의해서 스트리밍되기 때문입니다.
- 필드는 33가지를 설정할 수 있지만 필수적인 필드와 이후 다른 비즈니스 로직을 고려한 필드(리전별 코드) 몇 개 추가해줍니다.
- 참고로 샘플링 속도는 Kinesis Data Stream으로 전송되는 최종 사용자 요청 비율입니다. 모든 요청을 받고 싶으니 100%로 하겠습니다.
 
2. Kinesis 생성 및 구성

Amazon Kinesis 생성

 
- Kinesis를 생성함으로써 CloudFront에서 들어온 요청에 대한 로그를 실시간으로 스트리밍합니다.
- 해당 과정에서 샤드의 개수를 선택할 수 있는데 샤드 1개는 초당 1000MB를 처리할 수 있는 리소스이며 10개로 설정한 이유는 실제 로그의 데이터가 어느정도 일 지 알 수 없었기 때문입니다.
- 이후 확인한 결과 56만건의 요청이 수집되었을 때 1GB 가량이었기 때문에 사실상 샤드는 10개까지 필요하지 않다고 볼 수 있었습니다.
 
3. Firehose 생성 및 구성

Firehose 소스, 대상 설정

- Firehose는 스트리밍 데이터를 다양한 목적지로 로드하는 소비자 서비스입니다.
- 즉, Kinesis 스트림에 올라온 수집된 로그 데이터를 Amazon S3 로 업로드하는 역할을 합니다.

IAM 설정

버킷에 쌓인 로그

이런 과정을 거쳐 수집된 로그가 실시간으로 S3 버킷에 쌓이게 됩니다. 그럼 이제부터는 해당 로그들을 정제하여 보기 좋게 처리해야 합니다.
 
4. Athena 쿼리 작성 및 Glue 활용 테이블 정의

Athena 쿼리 테스트

 
우리가 정의한 방식으로 요청이 잘 들어왔는 지 상위 100개의 데이터를 표준 SQL을 통해 조회한 화면입니다. 이제 스케줄링을 통해 실제 데이터가 ElastiCache로 업데이트 될 수 있게 람다함수를 작성합니다.
 
5. 람다함수 작성

 

import boto3
import time
from datetime import datetime

def lambda_handler(event, context):
    athena_client = boto3.client('athena')
    
    # 현재 시간 가져오기
    now = datetime.utcnow()
    year = now.strftime('%Y')
    month = now.strftime('%m')
    day = now.strftime('%d')
    hour = now.strftime('%H')
    
    print(now)
    
    # Athena 쿼리 정의
    query = f"""
    WITH extracted_logs AS (
      SELECT 
        REGEXP_EXTRACT(cs_uri_stem, '/api/items/([0-9]+)', 1) AS item_number,
        from_unixtime(timestamp) AS log_datetime
      FROM logs
      WHERE cs_uri_stem LIKE '/api/items/%'
        AND from_unixtime(timestamp) BETWEEN (current_timestamp - interval '1' hour) AND current_timestamp
    )
    
    SELECT 
      item_number, 
      COUNT(*) AS request_count
    FROM extracted_logs
    WHERE date_format(log_datetime, '%Y') = '{year}'
      AND date_format(log_datetime, '%m') = '{month}'
      AND date_format(log_datetime, '%d') = '{day}'
      AND date_format(log_datetime, '%H') = '{hour}'
      AND date_format(log_datetime, '%i') = date_format(current_timestamp, '%i')
    GROUP BY item_number
    ORDER BY request_count DESC;
    """
    
    # Athena 쿼리 실행
    response = athena_client.start_query_execution(
        QueryString=query,
        QueryExecutionContext={
            'Database': 'default'  # 데이터베이스 이름
        },
        ResultConfiguration={
            'OutputLocation': 's3://get-ranknig/',
        }
    )
    
    query_execution_id = response['QueryExecutionId']
    print(f'QueryExecutionId: {query_execution_id}')
    
    # 쿼리 상태 확인
    status = 'RUNNING'
    while status in ['RUNNING', 'QUEUED']:
        time.sleep(3)  # 10초 대기
        result = athena_client.get_query_execution(QueryExecutionId=query_execution_id)
        status = result['QueryExecution']['Status']['State']
        print(f'Query status: {status}')
    
    if status == 'SUCCEEDED':
        print('Query succeeded')
    else:
        print(f'Query failed: {result["QueryExecution"]["Status"]["StateChangeReason"]}')
    
    return {
        'statusCode': 200,
        'body': f'Query execution status: {status}'
    }

(해당 코드는 사실 완성된 버전이 아닙니다.. 미처 최종 버전을 백업하지 못한 채 계정이 삭제되었기에..ㅜㅜ)
 
 

5. 최종 결과

최종 아키텍처

결과적으로 최종 아키텍처가 완성되었습니다!
 

분석된 로그

이처럼 분석된 결과가 5분에 하나씩 생성된 것을 볼 수 있습니다. 요청이 없을 때는 Redis에 저장된 데이터는 전부 빈 값으로 업데이트 되었던 이슈가 있어 이 부분은 Application 단에서 Default값을 주어 처리하였습니다.

CSV 확인

 

Kinesis 비용

다소 과한 샤드 10개를 사용했음에도 불구하고 DynamoDB와 비교하여 절반에 가까운 비용절감을 얻어냈고, 샤드를 적정개수인 1개로 설정하면 30분의 1 로 비용이 절약됩니다. Elasticache 비용은 이전 아키텍처와 동일하게 계산에 넣지 않았기 때문에 굳이 넣지 않았습니다.
 
 

5. 개선점

데이터 형식을 Parquet 형식으로 변환하여 분석에 최적화된 구조로 저장할 수 있다고 합니다. 적절한 테스트와 면밀한 요청 분석을 통해 샤드의 개수를 적절하게 선택할 수 있어야 하는 것이 개선해야 할 점입니다.