집 >백엔드 개발 >파이썬 튜토리얼 >Lambda 함수를 사용하여 So DynamoDB에서 데이터 구문 분석 및 로드

Lambda 함수를 사용하여 So DynamoDB에서 데이터 구문 분석 및 로드

Susan Sarandon원래의: 2025-01-06 06:24:47196검색

많은 시나리오에서는 JSON 형식의 데이터로 작업해야 하며, 데이터를 추출하고 처리한 다음 나중에 사용할 수 있도록 테이블에 저장하려고 합니다.

이 기사에서는 Lambda 함수를 사용하여 S3 버킷에서 DynamoDB 테이블로 JSON 형식의 데이터를 로드하는 방법에 대해 설명합니다

전제 조건

S3에 객체를 업로드할 권한이 있는 IAM 사용자
S3 및 DynamoDB에 대한 권한이 있는 Lambda 실행 역할

아키텍처 및 구성 요소

아래 아키텍처는 우리가 3가지 AWS 서비스를 사용하고 있음을 보여줍니다

S3 버킷
람다 함수
DynamoDB 테이블

Parsing & Loading Data from So DynamoDB with Lambda Function

아래의 다과 서비스에 대한 간략한 설명:

S3 버킷: 확장성, 보안성, 고성능 서비스를 갖춘 객체 스토리지 서비스가 데이터 스토리지 서비스로 사용됩니다
Lambda 기능: 인프라에 대한 걱정 없이 코드를 실행할 수 있고 설정이 쉬우며 다양한 프로그래밍 언어를 지원하는 서버리스 컴퓨팅 서비스입니다. 우리는 이를 활용하여 코드를 실행하고 로직을 배포할 것입니다.
DynamoDB: 데이터를 테이블에 저장하는 데 사용되는 서버리스 NoSQL 데이터베이스는 Lambda 함수로 처리된 데이터를 저장하는 데 사용됩니다.

흐름

사용자는 PutObject API 뒤에 있는 콘솔이나 CLI를 통해 JSON 파일을 S3 버킷에 업로드합니다
객체가 성공적으로 업로드되면 S3 이벤트가 트리거되어 람다 함수를 호출하여 파일을 로드하고 처리합니다
Lambda가 데이터를 처리하여 DynamoDB 테이블에 로드합니다

구현 단계

위 다이어그램 배포를 위한 단계 및 구성을 살펴보겠습니다

1- 아래 구성으로 Lambda 함수 생성

처음부터 저자
함수 이름: ParserDemo
런타임: Python 3.1x

나머지는 기본값으로 둡니다
Lambda가 생성된 후 아래와 같이 제한 시간 구성 및 실행 역할을 수정해야 합니다.

Parsing & Loading Data from So DynamoDB with Lambda Function

논리를 수행하기 위해 이 Python 코드를 작성했습니다

import json
import boto3

s3_client = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')

def lambda_handler(event, context):



    bucket_name = event['Records'][0]['s3']['bucket']['name'] # Getting the bucket name from the event triggered by S3
    object_key = event['Records'][0]['s3']['object']['key'] # Getting the Key of the item when the data is uploaded to S3
    print(f"Bucket: {bucket_name}, Key: {object_key}")


    response = s3_client.get_object(
    Bucket=bucket_name,
    Key=object_key
)


    # We will convert the streamed data into bytes
    json_data = response['Body'].read()
    string_formatted = json_data.decode('UTF-8') #Converting data into string

    dict_format_data = json.loads(string_formatted) #Converting Data into Dictionary 


    # Inserting Data Into DynamoDB

    table = dynamodb.Table('DemoTable')
    if isinstance(dict_format_data, list): #check if the file contains single record
        for record in dict_format_data:
            table.put_item(Item=record)

    elif isinstance(dict_format_data, dict): # check if the file contains multiple records 
        table.put_item(Item=data)

    else:  
        raise ValueError("Not Supported Format") # Raise error if nothing matched

2- S3 버킷 생성

버킷 이름: 고유한 이름을 사용하세요

나머지 구성은 기본값으로 둡니다

생성된 S3 버킷을 아래와 같이 람다 함수에 대한 트리거로 추가합니다.

Parsing & Loading Data from So DynamoDB with Lambda Function