따릉이 대여소 정보 Dashboard 구성(3) - Redshift Table Data 적재

AWS Cloud 실습을 위해 진행한 개인 프로젝트를 정리한 글입니다.

서울 공공데이터에서 따릉이 대여소 정보를 정제하여 CSV 파일로 만들었다.

만든 CSV 파일을 EC2 환경에서 Python Process를 통해 자동으로 AWS S3에 Upload 하였다.

AWS S3에 Object(CSV 파일)을 AWS Lambda를 이용해서 AWS Redshift Table에 적재하려 한다.

AWS Lambda Function 생성

S3 Bucket에 있는 CSV 파일을 Redshift Table에 적재하기 위해서 AWS Lambda를 이용하였다.

S3 Bucket에 새로운 Object가 Upload 할 때 해당 Object를 Redshift Table에 적재하도록 하였다.

S3 Object 관련된 블루프린트를 이용하여 Function을 생성한다.

해당 Function에 S3 권한 추가된 것을 확인한다.

S3 Object Upload Trigger를 매핑할 Bucket을 선택한다.

KMS Key 생성

AWS Lambda에서 환경변수를 암호화해줄 KMS(Key Management Service) Key를 생성한다.

대칭키로 구성해서 생성하였다.

AWS Lambda에서 [구성] - [권한]에서 역할을 확인한다.

KMS Key 생성 시 권한을 Lambda에서 확인 역할을 선택해준다.

Lambda 환경변수 Encryption

AWS Lambda Function에서 특정 값을 코드에 직접 입력하지 않고 환경변수로 넣었다.

환경변수를 KMS에서 만든 Key로 암호화를 하였다.

아래 5가지 값을 환경변수로 넣었다.

• Redshift JDBC Url
• Redshift 접속 유저 Password
• 적재할 Table명
• IAM Access Key
• IAM Secret Access Key

AWS Lambda에서 [구성] - [환경변수]에서 편집해서 추가해준다. 

환경 변수에 암호화 구성에서 도우미를 활성화한다.

AWS KMS에서 만든 Key를 선택해준다.

[보안 암호 코드 조각 해독]에서 복호화하는 코드를 확인한다.

AWS Lambda Function 코드

AWS Lambda Function Python 코드로 구성하였다.

Event에서 S3 Object명 Check

Lambda Function을 S3 Object Upload를 Trigger로 매핑하였다.

Function 첫 번째 Argument로 Event객체가 들어온다.

Event 객체에서 S3 Object명을 가져온다.

def lambda_handler(event, context):
    # Get the object from the event and show its content type
    s3_bucket = event['Records'][0]['s3']['bucket']['name']
    s3_object = event['Records'][0]['s3']['object']['key']

    s3_object = 's3://{}/{}'.format(s3_bucket, s3_object)

환경변수 복호화

위에서 설정한 환경변수를 복호화해준다.

kms = boto3.client('kms')
    
DBC = kms.decrypt(
	CiphertextBlob=b64decode(os.environ["CON"]),
	EncryptionContext={'LambdaFunctionName': os.environ['AWS_LAMBDA_FUNCTION_NAME']}
)['Plaintext'].decode('utf-8')

Redshift Connect / COPY

Python Alchemy package를 활용해서 Redshift에 접속하고 COPY 명령어로 S3 데이터를 옮겼다.

from sqlalchemy import create_engine 
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.engine.url import URL

url = URL.create(
	drivername='redshift+redshift_connector', # indicate redshift_connector driver and dialect will be used
    host=DBC, # Amazon Redshift host
    port=5439, # Amazon Redshift port
    database='dev', # Amazon Redshift database
    username='jsw', # Amazon Redshift username
    password=PASSWORD # Amazon Redshift password
)

# Connection
engine = create_engine(url)
db = scoped_session(sessionmaker(bind=engine))
    
copy_query = """COPY {} 
	FROM '{}' 
    CSV 
    DATAFORMAT AS 'MM/DD/YYYY' 
    IGNOREHEADER 1 
    CREDENTIALS 'aws_access_key_id={};aws_secret_access_key={}'
    DELIMITER ','""".format(RS_TABLE, s3_object, ACCESS_KEY, AWS_SECRET_ACCESS_KEY)
    
db.execute(copy_query)
db.commit()
db.close()

AWS Lambda Function Deploy

작성한 Code를 WEB Interface에서 바로 입력해도 된다.

여기는 AWS CLI를 통해서 Code를 Deploy 하였다.

먼저 기본적으로 제공하지 않는 Python Package를 Archive 해주었다.

그리고 Python Package를 Archive 한 zip에 Python Code를 넣어주었다.

# Python Package Path에 설치
mkdir package
pip3 install --target ./package sqlalchemy sqlalchemy-redshift redshift_connector

# Python Pakcage Archive
cd package
zip -r ../my-deployment-package.zip .

# Python Code 추가
cd ..
zip -g my-deployment-package.zip lambda_function.py

AWS CLI를 통해서 Python Package와 Code를 Deploy 한다.

aws lambda update-function-code --function-name seoulrentbike_csv_to_redshift --zip-file fileb://my-deployment-package.zip

AWS Lambda Function 확인

AWS Lambda는 CloudWatch에서 Log를 확인할 수 있다.

[모니터링] - [CloudWatch에서 로그 보기]에서 확인할 수 있다.

DBeaver Tool를 통해 Redshift에 접속하여 Data 적재가 잘 되었는지 확인한다.