如何借助AWS工具加速Python脚本批量处理邮箱地址并拆分任务至独立实例

实现方案（SQS + EC2自动缩放组） #

• 批量导入列表项：把所有待处理的列表项（如邮箱地址）逐条写入Amazon SQS标准队列，每条消息对应一个列表项。
• 配置EC2自动缩放组：
  • 创建EC2启动模板，预装Python环境和你的处理脚本，设置开机自动执行轮询SQS的脚本。
  • 配置自动缩放规则：根据SQS队列的待处理消息数自动增减EC2实例（比如消息数≥100时新增实例，≤10时缩减实例）。
• 实例端Python脚本逻辑：

  import boto3
  from your_script import process_item
  
  sqs_client = boto3.client('sqs')
  QUEUE_URL = "你的SQS队列URL"
  
  def poll_and_process():
      while True:
          # 长轮询获取消息，减少空轮询次数
          response = sqs_client.receive_message(
              QueueUrl=QUEUE_URL,
              MaxNumberOfMessages=1,
              WaitTimeSeconds=20
          )
  
          if "Messages" not in response:
              # 无消息时退出轮询（或根据需求保持监听）
              break
  
          for msg in response["Messages"]:
              item = msg["Body"]
              # 调用处理函数，传入列表项作为参数
              process_item(item)
              # 处理完成后删除消息，避免重复执行
              sqs_client.delete_message(
                  QueueUrl=QUEUE_URL,
                  ReceiptHandle=msg["ReceiptHandle"]
              )
  
  if __name__ == "__main__":
      poll_and_process()
  

• 触发执行：将所有列表项批量发送到SQS队列，自动缩放组会根据消息量启动对应数量的EC2实例并行处理。

具体实现步骤 #

1. 容器化邮箱处理脚本 #

• 编写Dockerfile打包脚本：

  FROM python:3.10-slim
  WORKDIR /app
  COPY requirements.txt .
  RUN pip install --no-cache-dir -r requirements.txt
  COPY email_processor.py .
  # 接受邮箱地址作为命令行参数
  ENTRYPOINT ["python", "email_processor.py"]
  

• 构建镜像并推送到Amazon ECR私有仓库。

2. 创建ECS Fargate任务定义 #

• 进入ECS控制台，创建Fargate类型的任务定义：
  • 配置CPU、内存规格（根据单邮箱处理的资源需求设置）。
  • 容器配置选择ECR中的镜像，预留命令参数位（后续由Step Functions传入邮箱地址）。
  • 配置IAM角色，确保任务能访问脚本依赖的AWS资源（如S3、数据库）。

3. 构建Step Functions状态机 #

• 用JSON定义状态机，核心使用Map状态遍历邮箱列表，为每个邮箱启动独立ECS任务：

  {
    "Comment": "分布式批量处理邮箱",
    "StartAt": "BatchProcessEmails",
    "States": {
      "BatchProcessEmails": {
        "Type": "Map",
        "ItemsPath": "$.emails",
        "MaxConcurrency": 60, // 根据AWS配额和成本调整并发数
        "Iterator": {
          "StartAt": "RunECSTask",
          "States": {
            "RunECSTask": {
              "Type": "Task",
              "Resource": "arn:aws:states:::ecs:runTask.sync",
              "Parameters": {
                "Cluster": "你的ECS集群名称",
                "TaskDefinition": "你的ECS任务定义ARN",
                "LaunchType": "FARGATE",
                "NetworkConfiguration": {
                  "AwsvpcConfiguration": {
                    "Subnets": ["子网ID-1", "子网ID-2"],
                    "SecurityGroups": ["安全组ID"],
                    "AssignPublicIp": "ENABLED"
                  }
                },
                "Overrides": {
                  "ContainerOverrides": [
                    {
                      "Name": "你的容器名称",
                      "Command": ["$$.Item"] // 传递当前遍历的邮箱地址
                    }
                  ]
                }
              },
              "End": true
            }
          }
        },
        "End": true
      }
    }
  }
  

• 为状态机配置IAM角色，授予调用ECS任务的权限。

4. 触发执行与监控 #

• 准备输入JSON：

  {
    "emails": ["user1@example.com", "user2@example.com", "..."]
  }
  

• 在Step Functions控制台启动状态机，传入上述输入，系统会自动为每个邮箱启动独立的Fargate容器处理。
• 通过CloudWatch监控ECS任务状态、状态机执行日志，排查失败任务；根据运行情况调整MaxConcurrency参数平衡效率与成本。

内容的提问来源于stack exchange，提问作者Graham Nedelka