(AWS자격증)AWS 관리형 서비스로 서버 없는 아키텍처 구현

서버없는 아키텍처 구현실습

AWS 람다실행

-

--

--------------------------

#-*- coding: utf-8 -*-

# TransactionProcessor Lambda 함수

#

# 이 함수는 Amazon S3 버킷에서 생성되는 객체가 트리거합니다.

# 파일이 다운로드되고 각 줄이 DynamoDB 테이블에 삽입됩니다.

from __future__ import print_function

import json, urllib, boto3, csv

# S3와 DynamoDB에 연결

s3 = boto3.resource('s3')

dynamodb = boto3.resource('dynamodb')

# DynamoDB 테이블에 연결

customerTable     = dynamodb.Table('Customer');

transactionsTable = dynamodb.Table('Transactions');

# Lambda 함수가 트리거될 때마다 실행되는 핸들러

def lambda_handler(event, context):

    # 디버깅 로그에 수신 이벤트 표시

    print("Event received by Lambda function: " + json.dumps(event, indent=2))

    # 이벤트에서 버킷과 객체 키 획득

    bucket = event['Records'][0]['s3']['bucket']['name']

    key = urllib.unquote_plus(event['Records'][0]['s3']['object']['key']).decode('utf8')

    localFilename = '/tmp/transactions.txt'

    # S3에서 로컬 파일 시스템으로 파일 다운로드

    try:

        s3.meta.client.download_file(bucket, key, localFilename)

    except Exception as e:

        print(e)

        print('Error getting object {} from bucket {}. 버킷이 존재하고이 버킷이 이 함수와 같은 리전에 있는지 확인하십시오. '.format (key, bucket))

        raise e

    # 트랜잭션 CSV 파일을 읽음 Delimiter is the '|' character

    with open(localFilename) as csvfile:

        reader = csv.DictReader(csvfile, delimiter='|')

        # 파일의 각 행을 읽음

        rowCount = 0

        for row in reader:

            rowCount += 1

            # 디버깅 로그에 행 표시

            print(row['customer_id'], row['customer_address'], row['trn_id'], row['trn_date'], row['trn_amount'])

            try:

                # 고객 ID와 주소를 고객 DynamoDB 테이블에 삽입

                customerTable.put_item(

                    Item={

                        'CustomerId': row['customer_id'],

                        'Address':    row['customer_address']})

                # 트랜잭션 세부 사항을 트랜잭션 DynamoDB 테이블에 삽입

                transactionsTable.put_item(

                    Item={

                        'CustomerId':        row['customer_id'],

                        'TransactionId':     row['trn_id'],

                        'TransactionDate':   row['trn_date'],

                        'TransactionAmount': int(row['trn_amount'])})

            except Exception as e:

                 print(e)

                 print("Unable to insert data into DynamoDB table".format(e))

        # 완료!

        return "%d transactions inserted" % rowCount

--------------------------

밑으로 내려가서

다시 위로 올라가서 S3추가

s3관련설정

----

람다 기능추가 

-----------------------

#-*- coding: utf-8 -*-

# TotalNotifier Lambda 함수

#

# 트랜잭션 DynamoDB 테이블에 값이 삽입되면 이 함수가 트리거됩니다.

# 트랜잭션 총계를 계산하고 한도를 초과하면 SNS로 알림을 보냅니다.

from __future__ import print_function

import json, boto3

# SNS에 연결

sns = boto3.client('sns')

alertTopic = 'HighBalanceAlert'

snsTopicArn = [t['TopicArn'] for t in sns.list_topics()['Topics'] if t['TopicArn'].endswith(':' + alertTopic)][0]

# DynamoDB에 연결

dynamodb = boto3.resource('dynamodb')

transactionTotalTableName = 'TransactionTotal'

transactionsTotalTable = dynamodb.Table(transactionTotalTableName);

# Lambda 함수가 트리거될 때마다 실행되는 핸들러

def lambda_handler(event, context):

    # 디버깅 로그에 수신 이벤트 표시

    print("Event received by Lambda function: " + json.dumps(event, indent=2))

    # 트랜잭션이 추가될 때마다 새 트랜잭션 총계 계산

    for record in event['Records']:

        customerId = record['dynamodb']['NewImage']['CustomerId']['S']

        transactionAmount = int(record['dynamodb']['NewImage']['TransactionAmount']['N'])

        # TransactionTotal DynamoDB 테이블에 고객 총계 업데이트

        response = transactionsTotalTable.update_item(

            Key={

                'CustomerId': customerId

            },

            UpdateExpression="add accountBalance :val",

            ExpressionAttributeValues={

                ':val': transactionAmount

            },

            ReturnValues="UPDATED_NEW"

        )

        # 최근 계정 잔액 검색

        latestAccountBalance = response['Attributes']['accountBalance']

        print("Latest account balance: " + format(latestAccountBalance))

        # 잔액이 1,500 USD를 초과하면 SNS로 메시지 전송

        if latestAccountBalance  >= 1500:

            # 전송할 메시지 구성

            message = '{"customerID": "' + customerId + '", ' + '"accountBalance": "' + str(latestAccountBalance) + '"}'

            print(message)

            # SNS로 메시지 전송

            sns.publish(

                TopicArn=snsTopicArn,

                Message=message,

                Subject='Warning! Account balance is very high',

                MessageStructure='raw'

            )

    # 완료!

    return 'Successfully processed {} records.'.format(len(event['Records']))

-------------------------------------------------------------------------------

Update total, send notification for balance exceeding $1500

-----------------------

밑에 내려가서 설정

DynamoDB추가

저장

SNS 만드릭

알람만들기

본인 이메일 확인

SQS 만들기

S3에 업로드 

---

추가

----------------------------

{

"Type" : "Notification",

"MessageId" : "eb0d030d-5f2d-5695-8f22-4c68d0335c0b",

"TopicArn" : "arn:aws:sns:us-east-1:123456789:HighAccountBalanceAlertSNSTopic",

"Subject" : "Warning! Account balance is very high",

"Message" : "{\\"customerID\\": \\"C2\\", \\"accountBalance\\":\\"1750\\"}",

----------------------------

실습확인 

-------------------

소스코드 예제.txt

-------------------

댓글과 공감은 필자에게 큰 힘이 됩니다.  ↓↓↓로그인 필요 없습니다↓↓↓  광고 클릭