[파이썬] AWS Lambda와 Kinesis를 통한 스트리밍 데이터 처리

05 Sep 2023

python

AWS Lambda와 Kinesis는 실시간 스트리밍 데이터 처리에 매우 효과적인 도구입니다. Lambda는 서버리스 컴퓨팅 플랫폼으로, 코드를 실행하는 데 필요한 서버 프로비저닝 및 관리 작업이 없어 편리합니다. Kinesis는 대량의 실시간 데이터를 처리하고 저장하는데 사용되는 스트리밍 데이터 플랫폼입니다.

이 블로그 포스트에서는 AWS Lambda와 Kinesis를 사용하여 스트리밍 데이터를 처리하는 과정을 다룹니다.

1. AWS Lambda 설정

먼저, AWS Lambda 함수를 생성해야 합니다. AWS 콘솔에서 Lambda 서비스로 이동하고 “함수 만들기”를 클릭합니다.

함수 생성 페이지에서 “함수 구성” 섹션에서 “처리 소스”로 “기본적으로 생성”을 선택합니다.
“추가 구성” 섹션에서 메모리, 실행 시간 등 함수의 구성을 설정합니다. 이는 함수의 성능과 비용에 영향을 미칩니다.
“함수 코드” 섹션에서 코드 작성 방법으로 “직접 업로드”를 선택하고, 업로드할 코드 파일을 선택합니다.

Lambda 함수는 이제 생성되었으며, 이제 Kinesis 스트림과 연결하여 데이터를 처리할 수 있습니다.

2. Kinesis 스트림 설정

Kinesis 스트림은 데이터를 스트리밍하는 데 사용되며, AWS 계정의 Kinesis 서비스에서 생성할 수 있습니다.

AWS Management Console에서 Kinesis 서비스로 이동한 다음, “스트림 만들기”를 클릭합니다.
스트림 이름을 설정하고, “스트림 생성”을 클릭합니다.

스트림이 생성되었으면, SDK 또는 API를 사용하여 생성한 Lambda 함수와 연결할 수 있습니다. Python을 사용하는 경우, boto3 라이브러리를 이용하여 Kinesis 스트림에 데이터를 보낼 수 있습니다.

3. Lambda 함수 코드 작성

이제 Lambda 함수를 작성하는 코드를 살펴보겠습니다. Python 언어를 사용하여 AWS Lambda 함수를 작성하는 방법은 매우 간단합니다.

import boto3
import json

def lambda_handler(event, context):
    # Kinesis 스트림 이름 설정
    stream_name = "your-stream-name"
    
    # Kinesis 클라이언트 생성
    kinesis = boto3.client("kinesis")
    
    # 이벤트 데이터 처리
    for record in event["Records"]:
        # 데이터 추출
        data = record["kinesis"]["data"]
        decoded_data = json.loads(data)
        
        # 데이터 처리 로직 수행
        # 여기에서는 단순히 데이터를 출력합니다
        print(decoded_data)
        
        # 데이터를 다른 Kinesis 스트림으로 전송할 수도 있습니다
        # kinesis.put_record(StreamName="your-other-stream-name", Data=json.dumps(decoded_data))

위 코드에서는 boto3 라이브러리를 사용하여 Kinesis 스트림과 연결하고, Lambda 함수의 이벤트(record)를 처리하는 간단한 예제를 보여줍니다.

이제 Lambda 함수가 Kinesis 스트림의 데이터를 소비하고 처리할 준비가 되었습니다.

4. Kinesis 스트림 데이터 전송

위에서 작성한 Lambda 함수를 통해 Kinesis 스트림으로 데이터를 보내는 방법은 여러 가지가 있습니다. 예를 들어, AWS SDK를 사용하여 데이터를 전송할 수도 있고, Kinesis producer library를 사용하여 대량의 데이터를 배치로 전송할 수도 있습니다.

다음은 Kinesis 스트림으로 데이터를 전송하는 예시 코드입니다.

import boto3

def put_record_to_kinesis(stream_name, data):
    kinesis = boto3.client("kinesis")
    response = kinesis.put_record(StreamName=stream_name, Data=data, PartitionKey="partition_key")
    print(response)

# 데이터 예시
data = '{"key": "value"}'

# 데이터 전송
put_record_to_kinesis("your-stream-name", data)

위 코드에서는 boto3 라이브러리를 사용하여 Kinesis 스트림으로 데이터를 전송하는 방법을 보여줍니다.

5. 결과 확인

Lambda 함수 및 Kinesis 스트림이 모두 설정되었으므로, 이제 실시간 스트리밍 데이터 처리의 결과를 확인할 수 있습니다. Lambda 함수에서 출력한 로그를 확인하거나 다른 Kinesis 스트림으로 데이터를 전송한 경우 해당 스트림에서 데이터를 소비할 수 있습니다.

이처럼, AWS Lambda와 Kinesis를 사용하면 대량의 실시간 스트리밍 데이터를 처리하고 저장하는 데 매우 유용한 솔루션을 제공합니다. Python을 사용하여 손쉽게 Lambda 함수를 작성하고 Kinesis와 통합할 수 있으므로, 스트리밍 데이터 처리에 이 도구를 활용해 보세요!