快速熟悉AWS Step Function

快速熟悉AWS Step Function

AWS Step Functions 将多个 AWS 服务协调为无服务器工作流，以便可以快速构建和更新应用程序。使用 Step Functions，可以设计和运行将AWS Lambda 和 Amazon ECS 等服务整合到功能丰富的应用程序中的工作流。工作流由一系列步骤组成，一个步骤的输出充当下一个步骤的输入。 使用 Step Functions，应用程序开发更简单、更直观，因为它将工作流转换为易于理解、易于向其他人说明且易于更改的状态机示意图。可以监控执行的每个步骤，这意味着可以快速识别并解决问题。Step Functions 可以自动触发和跟踪各个步骤，并在出现错误时重试，因此的应用程序能够按照预期顺序执行。

demo架构图

AWS step function 状态机的工作流程

工作流程如图：

1.调用 Input Lottery Winners 函数，传入 num_of_winners，进入第二步。

2.Random Select Winners 根据 Input Lottery Winners 的输出（body）调用 Random Select Winners，生成两个获奖号码，进入第三步。

3.Validate Winners 根据第二步输出查询 Winners 表判断是否重复，重复则传出 status：1，否则传出 status：0。

4.Is Winner In Past Draw 接受第三步 status 并判断，当 status 为1，则重新调用 Random Select Winners 进入第二步，当 status 为0，则调用 Notify Winners 和 Record Winner Queue，给 SNS topic 发送通知，把获奖者写入 Winner 表。

在整个工作流程中，当 Catch 接收到错误，都直接进入 Failed 步骤，输出异常并中断 step function。

创建步骤

IAM角色创建

执行 lambda 的角色需要以下策略：AmazonDynamoDBFullAccess

AWSLambdaBasicExecutionRole

AmazonSNSFullAccess

AWSStepFunctionsFullAccess

在 AWS IAM 控制台中创建角色：

大概如下图

lambda 创建

Input Lottery Winners

为了实现Step Functions状态机流转下的任务，我们这次实现会用到AWS Lambda作为我们业务的实现环境1.进入AWS控制台，选择服务然后输入Lambda进入AWS Lambda控制台

2.选择创建函数，然后选择从头开始创作来自定义我们的实验程序

3.首先我们需要创建状态机中的第一个状态任务Input Lottery Winners，输入函数名称Lottery-InputWinners来定义幸运儿的数量。运行语言选择Python 3.7。同时需要选择函数执行的权限, 这里我们选择使用现用角色，选择我们之前创建的IAM角色

5.在函数代码栏目下输入如下代码块，修改代码中的 region_name 为当前使用的 region

6.创建函数时复制页面右上角的ARN，后面创建状态机需要

import jsonclass CustomError(Exception):passdef lambda_handler(event, context):num_of_winners = event['input']

Trigger the Failed process

if 'exception' in event: raise CustomError("An error occurred!!") return { "body": { "num_of_winners": num_of_winners } }

接下来我们还需要创建另外三个需要定义的状态机业务逻辑，创建过程和上面的Lottery-InputWinners一致,下面是具体的状态机AWS

Lambda代码块

Lottery-RandomSelectWinners

import jsonimport boto3from random import randintfrom boto3.dynamodb.conditions import Key, AttrTOTAL_NUM = 10def lambda_handler(event, context):

variables

num_of_winners = event['num_of_winners'] # query in dynamodb dynamodb = boto3.resource('dynamodb', region_name='') table = dynamodb.Table('Lottery-Employee') # random select the winners, if has duplicate value, re-run the process while True: lottery_serials = [randint(1,TOTAL_NUM) for i in range(num_of_winners)] if len(lottery_serials) == len(set(lottery_serials)): break # retrieve the employee details from dynamodb results = [table.query(KeyConditionExpression=Key('lottery_serial').eq(serial), IndexName='lottery_serial-index') for serial in lottery_serials] # format results winner_details = [result['Items'][0] for result in results] return { "body": { "num_of_winners": num_of_winners, "winner_details": winner_details } }

Lottery-ValidateWinners

import json import boto3from boto3.dynamodb.conditions import Key, Attrdef lambda_handler(event, context):

variables

num_of_winners = event['num_of_winners'] winner_details = event['winner_details'] # query in dynamodb dynamodb = boto3.resource('dynamodb', region_name='') table = dynamodb.Table('Lottery-Winners') # valiate whether the winner has already been selected in the past draw winners_employee_id = [winner['employee_id'] for winner in winner_details] results = [table.query(KeyConditionExpression=Key('employee_id').eq(employee_id)) for employee_id in winners_employee_id] output = [result['Items'] for result in results if result['Count'] > 0] # if winner is in the past draw, return 0 else return 1 has_winner_in_queue = 1 if len(output) > 0 else 0 # format the winner details in sns winner_names = [winner['employee_name'] for winner in winner_details] name_s = "" for name in winner_names: name_s += name name_s += " " return { "body": { "num_of_winners": num_of_winners, "winner_details": winner_details }, "status": has_winner_in_queue, "sns": "Congrats! [{}] You have selected as the Lucky Champ!".format(name_s.strip()) }

Lottery-RecordWinners

import jsonimport boto3from boto3.dynamodb.conditions import Key, Attrdef lambda_handler(event, context):

variables

winner_details = event['winner_details'] # retrieve the winners' employee id employee_ids = [winner['employee_id'] for winner in winner_details] # save the records in dynamodb dynamodb = boto3.resource('dynamodb', region_name='') table = dynamodb.Table('Lottery-Winners') for employee_id in employee_ids: table.put_item(Item={ 'employee_id': employee_id }) return { "body": { "winners": winner_details }, "status_code": "SUCCESS" }

创建AWS SNS 通知服务

1.进入AWS控制台，在服务中搜索SNS

2.在SNS控制面板中，选择主题, 然后选择创建主题

3.在创建新主题弹框中，输入

主题名称: Lottery-Notification

显示名称: Lottery

1.创建主题后，会进入主题详细信息页面，这时候我们需要创建订阅来对接我们的消息服务，例如邮件服务（本次实验使用邮件服务来作为消息服务）

协议: Email

终端节点: <填入自己的邮箱地址>

2.复制主题详细页面的主题ARN,之后替换Step Functions状态机下的

创建Amazon Dynamodb 服务

本次实验需要创建两张Dynamodb表来记录员工信息和幸运儿信息。使用Dynamodb能更快地通过托管的方式记录数据同时免去数据库运维的压力。

1.进入AWS控制台，在服务中搜索Dynamodb

2.在左侧控制栏中选在表, 然后在主页面中选择创建表

3.在创建Dynamodb表中，填入如下信息

·表名称：Lottery-Winners ·主键：employee_id

·表名称：Lottery-Employee主键：employee_id

2.等待表创建完成后, 通过本附件中的request-items.json文件导入数据到Lottery-Employee

$ aws dynamodb batch-write-item --request-items file://request-items.json

主键：lottery_serial, 字段类型选择数字

索引名称：lottery_serial-index

创建AWS Step Functions 状态机

1.进入AWS控制台，在服务中搜索Step Functions

3.进入状态机主页面后，选择创建状态机

4.在定义状态机栏目下，选择默认使用代码段创作。同时在详细信息栏目输入状态机名称Lottery

5.在状态机定义栏目下，复制如下状态机定义文件，通过Amazon States Language来定义状态机的状态流转

{

"Comment": "A simple AWS Step Functions state machine that simulates the lottery session",

"StartAt": "Input Lottery Winners",

"States": {

"Input Lottery Winners": { "Type": "Task", "Resource": "", "ResultPath": "$", "Catch": [ { "ErrorEquals": [ "CustomError" ], "Next": "Failed" }, { "ErrorEquals": [ "States.ALL" ], "Next": "Failed" } ], "Next": "Random Select Winners" }, "Random Select Winners": { "Type": "Task", "InputPath": "$.body", "Resource": "", "Catch": [ { "ErrorEquals": [ "States.ALL" ], "Next": "Failed" } ], "Retry": [ { "ErrorEquals": [ "States.ALL"], "IntervalSeconds": 1, "MaxAttempts": 2 } ], "Next": "Validate Winners" }, "Validate Winners": { "Type": "Task", "InputPath": "$.body", "Resource": "", "Catch": [ { "ErrorEquals": [ "States.ALL" ], "Next": "Failed" } ], "Retry": [ { "ErrorEquals": [ "States.ALL"], "IntervalSeconds": 1, "MaxAttempts": 2 } ], "Next": "Is Winner In Past Draw" }, "Is Winner In Past Draw": { "Type" : "Choice", "Choices": [ { "Variable": "$.status", "NumericEquals": 0, "Next": "Send SNS and Record In Dynamodb" }, { "Variable": "$.status", "NumericEquals": 1, "Next": "Random Select Winners" } ] }, "Send SNS and Record In Dynamodb": { "Type": "Parallel", "End": true, "Catch": [ { "ErrorEquals": [ "States.ALL" ], "Next": "Failed" } ], "Retry": [ { "ErrorEquals": [ "States.ALL"], "IntervalSeconds": 1, "MaxAttempts": 2 } ], "Branches": [ { "StartAt": "Notify Winners", "States": { "Notify Winners": { "Type": "Task", "Resource": "arn:aws:states:::sns:publish", "Parameters": { "TopicArn": "", "Message.$": "$.sns" }, "End": true } } }, { "StartAt": "Record Winner Queue", "States": { "Record Winner Queue": { "Type": "Task", "InputPath": "$.body", "Resource": "", "TimeoutSeconds": 300, "End": true } } } ] }, "Failed": { "Type": "Fail" }

}

}

2.在配置设置下，选择为我创建IAM角色, 输入自定义的IAM角色名称MyStepFunctionsExecutionRole，并且附加AmazonSNSFullAccess权限

执行 Step Function 状态机1.入AWS控制台，在服务中搜索Step Functions

2.进入之前创建的状态机Lottery

4.在弹框中填入输入的json文本，这里的input代表在本次实验中需要抽取的获奖人数

实验结果1.Dynamodb表中Lottery-Winners记录获奖者

2.邮件会收取到幸运儿的信息

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。