processing-s3-uploads-with-step-functions

Name: processing-s3-uploads-with-step-functions
Author: aws

github.com/aws云与 DevOps3.3k5.1k 次安装1 天前更新

代码仓库

关于 processing-s3-uploads-with-step-functions

这是一项通过 AWS CLI 部署 AWS 工作流的技能，可按文件大小将 S3 上传任务路由到 Lambda 或 Fargate，并指导配置、监控和验证。

主要功能

01按文件大小路由
02响应 S3 对象创建事件
03用 Lambda 处理小文件
04用 Fargate 处理大文件
05配置监控与告警

适用场景

01处理大小不一的 S3 上传文件
02为大小文件选择不同计算服务
03部署文件处理 Step Functions 工作流

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/aws/agent-toolkit-for-aws --skill processing-s3-uploads-with-step-functions

安装后在 Claude Code 中输入/processing-s3-uploads-with-step-functions确认已加载。

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

references/

ecs-task-definition.md

iam-roles.md

scripts/

Dockerfile

ecs-trust-policy.json

eventbridge-trust-policy.json

fargate_processor.py

lambda_function.py

lambda-trust-policy.json

requirements.txt

statemachine.asl.json

stepfunctions-trust-policy.json

详细介绍

Step Functions 工作流：将 S3 上传路由到 Lambda 或 Fargate

概述

此技能使用 AWS CLI 部署事件驱动工作流。当文件上传到 S3 存储桶时，EventBridge 会触发 Step Functions 状态机。状态机会检查文件大小，并将处理任务路由到 Lambda 函数（文件 ≤ 6 MB）或 Fargate 任务（文件 > 6 MB）。

该架构包括：

启用了 EventBridge 通知的 S3 存储桶
在 S3 对象创建时触发 Step Functions 的 EventBridge 规则
使用 Choice 状态进行路由的 Step Functions 状态机
用于处理小文件的 Lambda 函数
用于处理大文件的 ECS Fargate 任务
包含两个子网、互联网网关和安全组的 VPC
用于存放 Fargate 容器镜像的 ECR 存储库
为 Lambda、Step Functions 和 ECS 任务设置的受限 IAM 角色

在以下情况下使用此技能：

你需要根据文件大小使用不同的计算资源处理 S3 上传
你想要一个能够处理小文件和大文件的无服务器工作流
你需要使用 Lambda 和 Fargate 编排 Step Functions

在以下情况下不要使用此技能：

所有文件都足够小，可以由 Lambda 处理（直接使用 S3 → Lambda）
你需要实时流处理（使用 Kinesis）
你不需要基于文件大小进行路由

先决条件

AWS CLI v2 — 已安装并配置。使用 aws sts get-caller-identity 验证。
Python 3.12 — 用于 Lambda 函数运行时。
Docker — 用于构建并推送 Fargate 容器镜像。

参数

bucket_name（必需）：S3 存储桶名称（全局唯一、小写，长度为 3-63 个字符）
region（必需）：所有资源所在的 AWS 区域
ecr_repo_name（必需）：ECR 存储库名称
state_machine_name（必需）：Step Functions 状态机名称
kms_key_arn（可选）：用于 CloudWatch Logs 加密的 KMS 密钥 ARN。如果未提供，请使用 aws kms create-key --description "Key for CloudWatch Logs encryption" --region {region} 创建一个

参数获取约束：

你必须在单个提示中一次性询问所有必需参数
你必须支持多种输入方式（直接输入、文件路径、URL）
继续操作前，你必须确认所有参数均已成功获取
你必须验证 bucket_name 是否符合 S3 命名规则

操作步骤

步骤 0：验证依赖项

约束：

你必须验证以下工具是否可用：aws-cli、python3（3.12+）、docker
发现任何缺失工具时，你必须向用户清楚说明
你必须询问用户是否要在缺少工具的情况下继续
你必须尊重客户在任何时候中止操作的决定
你必须向客户说明当前执行的步骤、执行原因以及将调用的工具

步骤 1：获取 AWS 账户 ID

约束：

你必须使用以下命令获取账户 ID：aws sts get-caller-identity --query 'Account' --output text
你必须将结果存储为 {account_id}，供后续所有步骤使用
如果未配置凭证，你必须中止操作

第 2 步：获取默认 VPC 和网络配置

约束：

必须使用以下命令获取默认 VPC ID： aws ec2 describe-vpcs --filters Name=isDefault,Values=true --query 'Vpcs[0].VpcId' --output text --region {region}
如果不存在默认 VPC，请告知用户必须使用 aws ec2 create-default-vpc --region {region} 创建一个，或者手动提供 VPC ID
必须从默认 VPC 获取两个子网 ID： aws ec2 describe-subnets --filters Name=vpc-id,Values={vpc_id} --query 'Subnets[0:2].SubnetId' --output text --region {region}
必须在默认 VPC 中创建安全组： aws ec2 create-security-group --group-name fargate-sg --description "Security group for Fargate tasks" --vpc-id {vpc_id} --region {region}
必须配置安全组的出站规则，仅允许 HTTPS 和 DNS 流量。首先撤销默认的允许所有流量出站规则： aws ec2 revoke-security-group-egress --group-id {sg_id} --ip-permissions IpProtocol=-1,IpRanges='[{CidrIp=0.0.0.0/0}]' --region {region} 然后添加范围受限的规则： aws ec2 authorize-security-group-egress --group-id {sg_id} --protocol tcp --port 443 --cidr 0.0.0.0/0 --region {region} 和 aws ec2 authorize-security-group-egress --group-id {sg_id} --protocol udp --port 53 --cidr 0.0.0.0/0 --region {region}
对于生产工作负载，必须建议使用 S3 和 CloudWatch Logs 的 VPC 端点，以避免流量经由互联网路由，并消除对宽泛出站规则的需求
必须记录 {vpc_id}、{subnet1_id}、{subnet2_id} 和 {sg_id}，供后续步骤使用

第 3 步：创建 ECR 存储库

约束：

必须使用以下命令创建存储库： aws ecr create-repository --repository-name {ecr_repo_name} --region {region}
必须从响应中记录 repositoryUri

第 4 步：构建并推送容器镜像

约束：

必须运行 docker --version，确认已安装 Docker。如果未安装 Docker，请指导用户从 https://docs.docker.com/get-docker/ 安装，并在 Docker 可用前中止操作
必须使用 ECR 对 Docker 进行身份验证： aws ecr get-login-password --region {region} | docker login --username AWS --password-stdin {account_id}.dkr.ecr.{region}.amazonaws.com
Dockerfile 和处理器代码位于 scripts/Dockerfile 和 scripts/fargate_processor.py

必须在 scripts 目录中构建并推送镜像：

cd scripts
docker build --platform linux/amd64 -t {ecr_repo_name} .
docker tag {ecr_repo_name}:latest {account_id}.dkr.ecr.{region}.amazonaws.com/{ecr_repo_name}:latest
docker push {account_id}.dkr.ecr.{region}.amazonaws.com/{ecr_repo_name}:latest
cd ..

第 5 步：创建 IAM 角色

按照 references/iam-roles.md 中的详细说明创建所有 IAM 角色（Lambda、ECS 任务执行、ECS 任务、Step Functions 和 EventBridge 角色）。

必须等待至少 10 秒，让 IAM 角色完成传播

第 6 步：创建 Lambda 函数

约束：

函数代码位于 scripts/lambda_function.py
打包和创建函数前，必须位于技能根目录
必须使用以下命令进行打包：python3 -c "import zipfile,io; z=io.BytesIO(); f=zipfile.ZipFile(z,'w'); f.writestr('lambda_function.py', open('scripts/lambda_function.py').read()); f.close(); open('/tmp/lambda_function.zip','wb').write(z.getvalue())"

必须使用以下命令创建函数：

aws lambda create-function \
    --function-name sfn-file-processor \
    --runtime python3.12 \
    --handler lambda_function.lambda_handler \
    --role arn:aws:iam::{account_id}:role/sfn-lambda-role \
    --zip-file fileb:///tmp/lambda_function.zip \
    --timeout 60 \
    --architectures x86_64 \
    --region {region}

必须使用以下命令验证函数已创建： aws lambda get-function --function-name sfn-file-processor --region {region}

第 7 步：创建 CloudWatch 日志组

约束：

必须使用以下命令为 Fargate 创建日志组： aws logs create-log-group --log-group-name /StepFunctionFargateTask --region {region}
必须使用 KMS 密钥加密日志组： aws logs associate-kms-key --log-group-name /StepFunctionFargateTask --kms-key-arn {kms_key_arn} --region {region}

第 8 步：创建 ECS 集群和任务定义

按照 references/ecs-task-definition.md 中的详细说明创建 ECS 集群并注册 Fargate 任务定义。

必须从响应中记录任务定义 ARN

第 9 步：创建启用 EventBridge 通知的 S3 存储桶

限制：

必须使用以下命令创建存储桶： aws s3api create-bucket --bucket {bucket_name} --region {region} --create-bucket-configuration LocationConstraint={region}
如果区域为 us-east-1，则不得包含 --create-bucket-configuration
必须在存储桶上启用 EventBridge 通知： aws s3api put-bucket-notification-configuration --bucket {bucket_name} --notification-configuration '{"EventBridgeConfiguration": {}}' --region {region}
必须在存储桶上启用默认加密： aws s3api put-bucket-encryption --bucket {bucket_name} --server-side-encryption-configuration '{"Rules":[{"ApplyServerSideEncryptionByDefault":{"SSEAlgorithm":"aws:kms"}}]}' --region {region}

第 10 步：创建 Step Functions 状态机

限制：

状态机定义位于 scripts/statemachine.asl.json

必须创建工作副本并替换所有占位符：

sed -e 's|${LambdaFunction}|arn:aws:lambda:{region}:{account_id}:function:sfn-file-processor|g' \
    -e 's|${Cluster}|arn:aws:ecs:{region}:{account_id}:cluster/sfn-cluster|g' \
    -e 's|${TaskDefinition}|{task_definition_arn}|g' \
    -e 's|${Subnet1}|{subnet1_id}|g' \
    -e 's|${Subnet2}|{subnet2_id}|g' \
    -e 's|${SecurityGroup}|{sg_id}|g' \
    scripts/statemachine.asl.json > /tmp/statemachine.asl.json

必须使用以下内容创建状态机：

aws stepfunctions create-state-machine \
    --name {state_machine_name} \
    --definition file:///tmp/statemachine.asl.json \
    --role-arn arn:aws:iam::{account_id}:role/sfn-state-machine-role \
    --type STANDARD \
    --region {region}

必须捕获响应中的 stateMachineArn

第 11 步：创建 EventBridge 规则

限制：

必须创建 EventBridge 规则，使其在创建 S3 对象时触发：

aws events put-rule \
    --name s3-to-stepfunctions \
    --event-pattern '{
      "source": ["aws.s3"],
      "detail-type": ["Object Created"],
      "detail": {
        "bucket": {
          "name": ["{bucket_name}"]
        }
      }
    }' \
    --region {region}

必须将状态机添加为目标：

aws events put-targets \
    --rule s3-to-stepfunctions \
    --targets '[{
      "Id": "StepFunctionsTarget",
      "Arn": "{state_machine_arn}",
      "RoleArn": "arn:aws:iam::{account_id}:role/sfn-eventbridge-role"
    }]' \
    --region {region}

第 12 步：配置监控

限制：

必须为失败的 EventBridge 调用创建死信队列： aws sqs create-queue --queue-name s3-to-stepfunctions-dlq --region {region}

必须更新 EventBridge 目标并附加死信队列：

aws events put-targets \
    --rule s3-to-stepfunctions \
    --targets '[{
      "Id": "StepFunctionsTarget",
      "Arn": "{state_machine_arn}",
      "RoleArn": "arn:aws:iam::{account_id}:role/sfn-eventbridge-role",
      "DeadLetterConfig": {
        "Arn": "arn:aws:sqs:{region}:{account_id}:s3-to-stepfunctions-dlq"
      }
    }]' \
    --region {region}

必须为 Step Functions 执行失败创建 CloudWatch 警报： aws cloudwatch put-metric-alarm --alarm-name sfn-execution-failures --metric-name ExecutionsFailed --namespace AWS/States --statistic Sum --period 300 --threshold 1 --comparison-operator GreaterThanOrEqualToThreshold --evaluation-periods 1 --dimensions Name=StateMachineArn,Value={state_machine_arn} --region {region}

第 13 步：验证

限制：

必须使用小文件（< 6 MB）进行测试，以验证 Lambda 处理：

echo 'test data' > /tmp/small-file.txt
aws s3 cp /tmp/small-file.txt s3://{bucket_name}/small-file.txt --region {region}

必须等待 15 秒，然后检查 Step Functions 执行情况： aws stepfunctions list-executions --state-machine-arn {state_machine_arn} --region {region}
必须确认执行成功并路由至 Lambda
必须汇总所有已创建的资源，包括：VPC ID、子网 ID、安全组 ID、ECR 存储库 URI、ECS 集群 ARN、任务定义 ARN、Lambda 函数 ARN、状态机 ARN、存储桶名称和 EventBridge 规则名称

故障排除

EventBridge 规则未触发

确认已在存储桶上启用 EventBridge 通知：aws s3api get-bucket-notification-configuration --bucket {bucket_name}
确认规则存在：aws events describe-rule --name s3-to-stepfunctions --region {region}
检查目标是否具有正确的状态机 ARN 和角色

Step Functions 执行在 Fargate 任务处失败

确认 ECR 中存在容器映像：aws ecr describe-images --repository-name {ecr_repo_name} --region {region}
检查子网是否可以访问互联网（路由表中需有 IGW）
确认安全组允许出站流量
检查 /StepFunctionFargateTask 中的 CloudWatch Logs

Lambda 调用失败

检查 CloudWatch Logs：aws logs tail /aws/lambda/sfn-file-processor --region {region}
确认 Step Functions 角色具有 lambda:InvokeFunction 权限

IAM PassRole 错误

Step Functions 角色必须对 ECS 执行角色和任务角色 ARN 都具有 iam:PassRole 权限

Fargate 任务卡在 PROVISIONING 状态

确认子网已启用自动分配公有 IP
确认已附加互联网网关，且路由表具有 0.0.0.0/0 路由

安全注意事项

带有公有 IP 的 Fargate 任务会暴露在互联网上。撤销默认的允许所有出站流量规则，并配置范围受限的出站规则：aws ec2 revoke-security-group-egress --group-id {sg_id} --ip-permissions IpProtocol=-1,IpRanges='[{CidrIp=0.0.0.0/0}]'，然后添加 aws ec2 authorize-security-group-egress --group-id {sg_id} --protocol tcp --port 443 --cidr 0.0.0.0/0 和 aws ec2 authorize-security-group-egress --group-id {sg_id} --protocol udp --port 53 --cidr 0.0.0.0/0。在生产环境中，可考虑使用 VPC 端点访问 S3 和 CloudWatch Logs，而不是通过互联网路由流量。
将容器镜像推送到 ECR 前，先扫描漏洞。使用以下命令启用 ECR 镜像扫描：aws ecr put-image-scanning-configuration --repository-name {ecr_repo_name} --image-scanning-configuration scanOnPush=true --region {region}
使用 IAM 角色管理凭证，切勿在容器代码中硬编码访问密钥。
为 S3 存储桶启用静态加密：aws s3api put-bucket-encryption --bucket {bucket_name} --server-side-encryption-configuration '{"Rules":[{"ApplyServerSideEncryptionByDefault":{"SSEAlgorithm":"aws:kms"}}]}'
为 Fargate 容器日志启用 CloudWatch Logs 加密：aws logs associate-kms-key --log-group-name /StepFunctionFargateTask --kms-key-arn <KMS_KEY_ARN>
在 EventBridge 规则上为调用失败配置死信队列。
为 Step Functions 执行失败设置 CloudWatch 警报，以便了解运行情况。

版本信息

AWS CLI：2.x
Python 运行时：3.12
上次验证时间：2026-04-27

基本信息

作者: aws
仓库: aws/agent-toolkit-for-aws
分类: 云与 DevOps
安装次数: 5,131
GitHub Stars: 3,330

报告问题