aws(学习笔记第二十四课) 使用sam开发step functions

🕗 发布于 2025-01-14 20:45 aws 学习笔记

aws(学习笔记第二十四课)

使用sam开发step functions

学习内容：

生成sam的step functions实例程序
什么是SAM amazon Serverless Application Model
SAM程序结构
SAM执行程序

1. 生成`sam`的`step functions`实例程序

参照文档
这里参照AWS的官方文档SAM amazon Serverless Application Model
什么是SAM amazon Serverless Application Model
- 整体架构
  SAM就是一个基于Cloudformation的应用程序框架，主要目的正如名字(Serverless Application Model)，方便进行Serverless Application的开发。
  开发的一般步骤如下：
  - 开发lambda等serverless application。
  - 利用step functions等serverless application。当然，其中可以调用lambda。
  - 之后利用上面的serverless application，进行Cloudformation的template定义。
  - 最后经过sam build和sam deploy部署到AWS的环境中。

3. `SAM`程序结构

开始使用SAM

这里使用SAM提供的实例程序进行练习
- HourlyTradingSchedule是一个AWS EventBridge的规则，这里定义了股票投资程序的调用周期。类似于linux的cron job。
- StockTradingStateMachine就是股票交易的StateMachine。
- 股票交易的StateMachine里面包括三个lambda
  - StockCheckerFunction这里随机产生股票的价格（进行简单的股票市场的模拟）
  - 中间其实有一个choice state，进行判断。这里没有画出来
  - 之后根据判断，如果股票高过某个固定价格，那么进行StockBuyerFunction的调用
  - 如果股票高过某个固定价格，那么进行StockSellerFunction的调用
  - 最后，不管买还是卖的操作，都进行TransactionTable的写入（使用DynamoDB记录交易）
进行实际代码的实验
- 实验环境
  这里还是使用非常给力的工具CloudShell。
- 构建代码
  - 创建代码的父路径
```
mkdir demo-sam
cd demo-sam
```
  - 使用sam生成股票实例代码（这个代码是sam自带的）
    
    之后进行一些runtime的相关设定。
    
    到这里，代码就会被生成出来，而且cloudformation的template文件都是yaml格式的。

将cloudshell环境中的代码通过S3取到本地

创建传输文件的S3 bucket
因为cloudshell不是很容易和本地传输文件，所以使用S3 bucket。

将SAM代码打包，copy到S3 bucket

tar zcvf demo-sam.tar.gz demo-sam/
aws s3 cp demo-sam.tar.gz s3://finlay-cloudshell/

将SAM init生成的实例程序代码，下载到本地

本地文件夹如下所示。

将SAM init生成的实例程序代码使用vscode打开（这里单纯的可以更加容易编辑代码）

template文件

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: |
  demo-sam
  Sample SAM Template for demo-sam

Resources:
  StockTradingStateMachine:
    Type: AWS::Serverless::StateMachine # More info about State Machine Resource: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-statemachine.html
    Properties:
      DefinitionSubstitutions:
        StockCheckerFunctionArn: !GetAtt StockCheckerFunction.Arn
        StockSellerFunctionArn: !GetAtt StockSellerFunction.Arn
        StockBuyerFunctionArn: !GetAtt StockBuyerFunction.Arn
        DDBPutItem: !Sub arn:${AWS::Partition}:states:::dynamodb:putItem
        DDBTable: !Ref TransactionTable
      Events:
        HourlyTradingSchedule:
          Type: Schedule # More info about Schedule Event Source: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-property-statemachine-schedule.html
          Properties:
            Description: Schedule to run the stock trading state machine every hour
            Enabled: false # This schedule is disabled by default to avoid incurring charges.
            Schedule: rate(1 hour)
      Policies:
        # Find out more about SAM policy templates: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-policy-templates.html
        - LambdaInvokePolicy:
            FunctionName: !Ref StockCheckerFunction
        - LambdaInvokePolicy:
            FunctionName: !Ref StockSellerFunction
        - LambdaInvokePolicy:
            FunctionName: !Ref StockBuyerFunction
        - DynamoDBWritePolicy:
            TableName: !Ref TransactionTable
      DefinitionUri: statemachine/stock_trader.asl.json

  StockCheckerFunction:
    Type: AWS::Serverless::Function # More info about Function Resource: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-function.html
    Properties:
      CodeUri: functions/stock_checker/
      Handler: app.lambda_handler
      Runtime: python3.9
      Architectures:
        - x86_64

  StockSellerFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: functions/stock_seller/
      Handler: app.lambda_handler
      Runtime: python3.9
      Architectures:
        - x86_64

  StockBuyerFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: functions/stock_buyer/
      Handler: app.lambda_handler
      Runtime: python3.9
      Architectures:
        - x86_64

  TransactionTable:
    Type: AWS::Serverless::SimpleTable # More info about SimpleTable Resource: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-simpletable.html
    Properties:
      PrimaryKey:
        Name: Id
        Type: String
      ProvisionedThroughput:
        ReadCapacityUnits: 1
        WriteCapacityUnits: 1

  ApplicationResourceGroup:
    Type: AWS::ResourceGroups::Group
    Properties:
      Name: !Sub ApplicationInsights-SAM-${AWS::StackName}
      ResourceQuery:
        Type: CLOUDFORMATION_STACK_1_0
  ApplicationInsightsMonitoring:
    Type: AWS::ApplicationInsights::Application
    Properties:
      ResourceGroupName: !Ref ApplicationResourceGroup
      AutoConfigurationEnabled: 'true'
Outputs:
  # StockTradingStateMachineHourlyTradingSchedule is an implicit Schedule event rule created out of Events key under Serverless::StateMachine
  # Find out more about other implicit resources you can reference within SAM
  # https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-specification-generated-resources.html
  StockTradingStateMachineArn:
    Description: Stock Trading State machine ARN
    Value: !Ref StockTradingStateMachine
  StockTradingStateMachineRoleArn:
    Description: IAM Role created for Stock Trading State machine based on the
      specified SAM Policy Templates
    Value: !GetAtt StockTradingStateMachineRole.Arn

使用vscode打开template文件，可以看到AWS service的图形化结构。
在这里插入图片描述

stock_checker文件 随机产生stock的价格

from random import randint


def lambda_handler(event, context):
    """Sample Lambda function which mocks the operation of checking the current price 
    of a stock.

    For demonstration purposes this Lambda function simply returns 
    a random integer between 0 and 100 as the stock price.

    Parameters
    ----------
    event: dict, required
        Input event to the Lambda function

    context: object, required
        Lambda Context runtime methods and attributes

    Returns
    ------
        dict: Object containing the current price of the stock
    """
    # Check current price of the stock
    stock_price = randint(
        0, 100
    )  # Current stock price is mocked as a random integer between 0 and 100
    return {"stock_price": stock_price}

stock_buyer文件 如果价格低，选择买进

from datetime import datetime
from random import randint
from uuid import uuid4


def lambda_handler(event, context):
    """Sample Lambda function which mocks the operation of buying a random number
    of shares for a stock.

    For demonstration purposes, this Lambda function does not actually perform any 
    actual transactions. It simply returns a mocked result.

    Parameters
    ----------
    event: dict, required
        Input event to the Lambda function

    context: object, required
        Lambda Context runtime methods and attributes

    Returns
    ------
        dict: Object containing details of the stock buying transaction
    """
    # Get the price of the stock provided as input
    stock_price = event["stock_price"]
    # Mocked result of a stock buying transaction
    transaction_result = {
        "id": str(uuid4()),  # Unique ID for the transaction
        "price": str(stock_price),  # Price of each share
        "type": "buy",  # Type of transaction (buy/sell)
        "qty": str(
            randint(1, 10)
        ),  # Number of shares bought/sold (We are mocking this as a random integer between 1 and 10)
        "timestamp": datetime.now().isoformat(),  # Timestamp of the when the transaction was completed
    }
    return transaction_result

stock_seller文件 如果价格高，选择抛出

from datetime import datetime
from random import randint
from uuid import uuid4


def lambda_handler(event, context):
    """Sample Lambda function which mocks the operation of selling a random number
    of shares for a stock.

    For demonstration purposes, this Lambda function does not actually perform any 
    actual transactions. It simply returns a mocked result.

    Parameters
    ----------
    event: dict, required
        Input event to the Lambda function

    context: object, required
        Lambda Context runtime methods and attributes

    Returns
    ------
        dict: Object containing details of the stock selling transaction
    """
    # Get the price of the stock provided as input
    stock_price = event["stock_price"]
    # Mocked result of a stock selling transaction
    transaction_result = {
        "id": str(uuid4()),  # Unique ID for the transaction
        "price": str(stock_price),  # Price of each share
        "type": "sell",  # Type of transaction (buy/sell)
        "qty": str(
            randint(1, 10)
        ),  # Number of shares bought/sold (We are mocking this as a random integer between 1 and 10)
        "timestamp": datetime.now().isoformat(),  # Timestamp of the when the transaction was completed
    }
    return transaction_result

step functions文件 将上面的lambda函数穿插起来，形成一个工作流程编排

{
    "Comment": "A state machine that does mock stock trading.",
    "StartAt": "Check Stock Value",
    "States": {
        "Check Stock Value": {
            "Type": "Task",
            "Resource": "${StockCheckerFunctionArn}",
            "Retry": [
                {
                    "ErrorEquals": [
                        "States.TaskFailed"
                    ],
                    "IntervalSeconds": 15,
                    "MaxAttempts": 5,
                    "BackoffRate": 1.5
                }
            ],
            "Next": "Buy or Sell?"
        },
        "Buy or Sell?": {
            "Type": "Choice",
            "Choices": [
                {
                    "Variable": "$.stock_price",
                    "NumericLessThanEquals": 50,
                    "Next": "Buy Stock"
                }
            ],
            "Default": "Sell Stock"
        },
        "Sell Stock": {
            "Type": "Task",
            "Resource": "${StockSellerFunctionArn}",
            "Retry": [
                {
                    "ErrorEquals": [
                        "States.TaskFailed"
                    ],
                    "IntervalSeconds": 2,
                    "MaxAttempts": 3,
                    "BackoffRate": 1
                }
            ],
            "Next": "Record Transaction"
        },
        "Buy Stock": {
            "Type": "Task",
            "Resource": "${StockBuyerFunctionArn}",
            "Retry": [
                {
                    "ErrorEquals": [
                        "States.TaskFailed"
                    ],
                    "IntervalSeconds": 2,
                    "MaxAttempts": 3,
                    "BackoffRate": 1
                }
            ],
            "Next": "Record Transaction"
        },
        "Record Transaction": {
            "Type": "Task",
            "Resource": "${DDBPutItem}",
            "Parameters": {
                "TableName": "${DDBTable}",
                "Item": {
                    "Id": {
                        "S.$": "$.id"
                    },
                    "Type": {
                        "S.$": "$.type"
                    },
                    "Price": {
                        "N.$": "$.price"
                    },
                    "Quantity": {
                        "N.$": "$.qty"
                    },
                    "Timestamp": {
                        "S.$": "$.timestamp"
                    }
                }
            },
            "Retry": [
                {
                    "ErrorEquals": [
                        "States.TaskFailed"
                    ],
                    "IntervalSeconds": 20,
                    "MaxAttempts": 5,
                    "BackoffRate": 10
                }
            ],
            "End": true
        }
    }
}

检查vscode的sam预览功能
在这里插入图片描述

回到cloudshell执行sam build和sam deploy部署到AWS
```
sam build
sam deploy
```
可以看到，本质上这个sam application还是使用Cloudformation进行部署。

查看Cloudformation进一步验证了想定结果。

4. 执行`sam`的`step functions`实例程序

打开默认的aws eventbridge的规则设置
默认是禁用的，编辑这里，打开禁用。
查看执行结果
可以看出，已经成功执行一次

股票价格为2，执行了buy stock lambda

原文地址：https://blog.csdn.net/sealaugh1980/article/details/145092765

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：【机器学习：十三、PyTorch简介及实现】
下一篇：[NOIP2007 提高组] 矩阵取数游戏

论文阅读：SplatMAP: Online Dense Monocular SLAM with 3D Gaussian Splatting
1 Introduction为了实现具有高保真渲染的实时三维重建，研究人员已经探索了将SLAM与可微渲染相结合的潜力。然而，单目SLAM系统，特别是在早期建图阶段，由于观测有限、三角测量基线较浅以及位
阅读更多2025-01-15
B3DM转换成FBX
B3DM（Binary 3D Model）是一种用于存储三维模型的二进制格式，特别适用于大规模的三维城市建模和地理信息系统（GIS）应用‌‌。B3DM文件的设计目的是为了高效地存储和传输包含几何数据、
阅读更多2025-01-15
字符串算法篇——字里乾坤，算法织梦，解构字符串的艺术（下）
上篇我们介绍了常用的字符串及其相关原理应用，本篇将结合具体题目，进一步深化对于字符串算法的掌握运用。
阅读更多2025-01-15
intel x99主板设置上电服务器自动启动
Bootup Numlock State它决定了在计算机开机时，数字小键盘（Num Lock）是处于开启还是关闭状态。当设置为开启（On）时，计算机启动后数字小键盘就可以直接输入数字；如果设置为关闭（
阅读更多2025-01-15
语音识别的预训练模型
Whisper 是一个强大的用于语音识别和翻译的预训练模型。与 Wav2Vec2 相比，它具有更高的转写准确性，输出包含标点和大小写。它可用于转写英语和其他 96 种语言的语音，既能处理短音频片段又能
阅读更多2025-01-15
Python进制转换
数据范围保证：2≤N,M≤162≤N,M≤16，若 N≥10N≥10，则用 A∼FA∼F 表示字码 10∼1510∼15。保证 SS 对应的十进制数的位数不超过 1010。第一行为一个整数 TT，表示
阅读更多2025-01-15
【已解决】git clone报错：Failed to connect to github.com port 443: Timed out
因为git在拉取或者提交项目时，中间会有git的http和https代理，但是我们本地环境本身就有SSL协议了，所以取消git的https代理即可，不行再取消http的代理。
阅读更多2025-01-15
大数据技术Kafka详解 ⑤ | Kafka中的CAP机制
本文详细介绍Kafka分布式系统的三个指标与机制。
阅读更多2025-01-15
ArkTS 组件事件、状态管理与资源管理
凡是过去，皆为序章；凡是未来，皆有可期。
阅读更多2025-01-15
lammps应用于热电材料
本文介绍了LAMMPS在热电材料中的应用，介绍了热传导、傅里叶定律、非平衡分子动力学等理论基础，举例计算了Ar、SiGe等材料的热导率
阅读更多2025-01-15

aws(学习笔记第二十四课) 使用sam开发step functions