亚马逊批处理作业处理队列

Question

我正在使用 AWS EC2 实例进行生物信息学工作。我有一些（约 1000 个）大文件，应该使用 EC2 实例上的脚本进行处理，结果应该上传回 S3 存储桶。我想将作业（文件）分配给多个 EC2 实例，优先以现货价格开始。

我需要的是一个简单易用的排队系统（可能是 AWS SQS 或其他），它可以将作业分配给实例并在实例失败时重新启动作业（由于现货价格过高或其他原因）。我研究过 AWS SQS 示例，但这些示例太高级了，通常涉及自动缩放和复杂的消息生成应用程序。

有人可以从概念上指出如何以最好和最简单的方式解决这个问题吗？这个简单的 AWS SQS 应用程序有什么例子吗？一堆实例应该如何启动，如何告诉然后监听队列？

对于每个输入文件，我的工作流程基本上是这样的：

aws s3 cp s3://mybucket/file localFile ## Possibly streaming the file without copy
work.py --input localFile --output outputFile
aws s3 cp outputFile s3://mybucket/output/outputFile

Answer 1

截至 2016 年 12 月，AWS 推出了一项名为 AWS Batch 的服务 对于问题中描述的工作量，这可能很好（甚至可能很好）。请在选择其他建议之一之前查看 Batch。

Answer 2

您正在描述一种非常常见的面向批处理的设计模式：

• 工作被放置在一个队列中
• 一个或多个“工作”实例拉从队列中工作
• 工作人员实例的数量根据队列的大小和工作的紧迫性扩展
• 使用现货定价最大限度地降低成本

最好的方法是：

• 使用 Amazon Simple Queuing Service (SQS) 存储工作请求
• 启动 Amazon EC2 实例，每个实例都重复：
  • 从队列中拉出一条消息
  • 处理消息（例如通过您上面列出的下载/处理/上传步骤）
  • 删除队列中的消息（表示工作已完成）
• 使用Auto-Scaling控制实例数量，这样积压大的地方可以启动更多的实例，没有工作的时候可以关闭所有的实例
• 将 Spot 定价 与 Auto-Scaling 组结合使用，以便在 Spot 价格降低您的最高出价时实例会自动“恢复活力”

与其让排队系统“将作业分配给实例并在实例失败时重新启动作业”，SQS 将仅用于存储作业。 Auto-Scaling 将负责启动实例（包括重启是 Spot 价格变化），并且实例本身将从队列中拉出工作。将其视为“拉”模型而不是“推”模型。

虽然整个系统可能看起来很复杂，但每个单独的组件都非常简单。我建议一步一步做：

1. 拥有一个以某种方式将工作请求推入 SQS 队列的系统。这可以像使用 CLI 中的 aws sqs put-message 或在 Python using Boto（适用于 Python 的 AWS 开发工具包）中添加几行代码一样简单。

这里有一些示例代码（在命令行中通过消息调用它）：

#!/usr/bin/python27

import boto, boto.sqs
from boto.sqs.message import Message
from optparse import OptionParser

# Parse command line
parser = OptionParser()
(options, args) = parser.parse_args()

# Send to SQS
q_conn = boto.sqs.connect_to_region('ap-southeast-2')

q = q_conn.get_queue('my-queue')
m = Message()
m.set_body(args[0])
print q.write(m)

print args[0] + ' pushed to Queue'

1. 配置一个 Amazon EC2 实例，它可以自动启动您的 Python 应用程序或从 SQS 中提取并处理您的工作的脚本。使用User Data 字段在实例启动时触发工作。从 shell 脚本运行您的工作流程，或者您也可以将 S3 上传/下载代码编写为 Python 应用程序的一部分（包括不断拉取新消息的循环）。

下面是一些从队列中检索消息的代码：

#!/usr/bin/python27

import boto, boto.sqs
from boto.sqs.message import Message

# Connect to Queue
q_conn = boto.sqs.connect_to_region('ap-southeast-2')
q = q_conn.get_queue('my-queue')

# Get a message
m = q.read(visibility_timeout=15)
if m == None:
  print "No message!"
else:
  print m.get_body()
  q.delete_message(m)

1. 配置与您刚刚为 EC2 创建的配置相匹配的 Auto-Scaling 启动配置。这会告诉 Auto-Scaling 如何启动实例（例如实例类型、用户数据）以及您愿意支付的 Spot 价格。
2. 创建 Auto Scaling 组以自动启动实例。
3. 如果您希望 Auto Scaling 组根据队列大小添加/删除实例，请配置 Scaling 策略

另见：

• AWS Reference Architectures 上的批处理
• Cloud Design Pattern: Job Observer
• 文章：Auto-scaling Amazon EC2 with Amazon SQS
• 文章：Getting Started with AWS and Python
• 自定进度实验室：Launching Amazon EC2 Spot Instances with Auto Scaling and Amazon CloudWatch（使用命令行而不是控制台的 Auto Scaling，因此有点复杂）
• GitHub：SampleBatchProcessing
• Using SQS with Python and boto