Lambda : s3 csv 파일을 pandas로 변환하고 자동으로 Dynamo에 저장

소개

간단한 내용입니다만, Lambda에서의 pandas의 기동이나 csv 로딩, DynamoDB 격납을 위한 float⇒Decimal 변환등, 일부 끼워넣어 포인트가 있었으므로, 비망록도 겸해 기재합니다.

하고 싶은 일

s3에 csv 파일을 업로드하면 자동으로 DynamoDB에 저장합니다.

s3 ⇒ Lambda ⇒ DynamoDB

DynamoDB 만들기

DynamoDB를 만듭니다.
기본 키는 일단 색인화됩니다.
정렬 키 추가로 확장할 수 있습니다.

Lambda 함수 만들기(설정)

Lambda 함수를 만듭니다.
역할에는 s3 및 DynamoDB에 액세스할 수 있는 권한을 둡니다.
리전은 s3과 일치합니다.



Lambda 함수의 트리거를 설정합니다.

Lambda 함수 만들기(코드)

Lambda_function.py

import boto3
import json
import io
import decimal
import pandas as pd

from decimal import *
getcontext().prec = 6

dynamodb = boto3.resource('dynamodb')

# convert function to store data to dynamodb.
def convert_float_to_Decimal(dictionary):
    for key in dictionary:
        if isinstance(dictionary[key], float):
            dictionary[key] = Decimal(str(dictionary[key]))
    return dictionary

def insert(index, dictionary):
    dictionary["index"] = int(index)
    import_item = convert_float_to_Decimal(dictionary)
    table = dynamodb.Table('test')
    table.put_item(Item=import_item)

# read csv from s3 and insert data as dictionary to dynamodb.
def lambda_handler(event, context):
    client = boto3.client('s3')
    # retrieve directory from event.
    key = event['Records'][0]['s3']['object']['key']
    obj = client.get_object(Bucket='dataset-pract',Key=key)
    initial_df = pd.read_csv(io.BytesIO(obj['Body'].read()),
    index_col=0)
    initial_df_index = list(initial_df.index)
    # rewrite df as list of dict.
    initial_df = initial_df.to_dict("records")
    for index, dictionary in zip(initial_df_index, initial_df):
        insert(index, dictionary)

float 형을 Decimal 형으로 변환

DynamoDB에는 float 형식으로 데이터를 넣을 수 없으므로 저장하고 싶은 사전 형식 모든 key에 대해 value가 float이면 Decimal로 변환하고 있습니다.
변환시, float로부터 직접 Decimal로 변환하면 잘되지 않기 때문에, 일단 string를 끼웁니다.
htps : // 기주 b. 코 m / 보토 / 보토 3 / 이스에 s / 665

CSV 로딩

get_object로 지정된 버킷에서 읽습니다. 그런 다음 Body를 확장하여 바이트 데이터를 변환하고 read_csv로 읽습니다.

Pandas import 정보

Lambda에서 Pandas를 사용하려면 패키지를 설치해야합니다.
이 경우, Pandas의 폴더와 Lambda 함수를 zip화해 업로드하면 됩니다만, Pandas의 임포트에는 의존관계가 있으므로 아래 사이트로부터 다운로드한 파일을 사용하는 것이 편합니다.

Lambda용 Pandas



이렇게 다운로드한 폴더에 작성한 Lambda 함수를 넣어 zip화합니다.

zip 업로드



트리거 시작 시 실행할 함수를 지정합니다.
이번에는 Lambda_handler로 만들겠습니다.

업로드하고 저장하면 완료됩니다.

테스트

사용할 데이터

import pandas as pd
data = pd.read_csv("USAccDeaths.csv", index_col=0)
print(data.head())

이것을 s3의 지정된 버킷에 넣습니다.

DB를 확인합니다.



안전하게 들어갔습니다.

이상입니다.