AWS Batch

这包含 TorchX 的 AWS Batch 调度程序，可以用于在 AWS Batch 上直接运行 TorchX 组件。

此调度程序处于原型阶段，可能会在未通知的情况下发生变化。

预备知识

您需要创建一个配置为多节点并行作业的 AWS Batch 队列。

查看 https://docs.aws.amazon.com/batch/latest/userguide/Batch_GetStarted.html#first-run-step-2 以了解如何设置工作队列和计算环境。它需要由EC2支持，以便进行多节点并行任务。

参见 https://docs.aws.amazon.com/batch/latest/userguide/multi-node-parallel-jobs.html 以获取有关分布式作业的更多信息。

如果你要使用工作区和容器补丁功能，你还需要配置一个 Docker 注册表来存储包含你更改的补丁容器，例如 AWS ECR。

参见 https://docs.aws.amazon.com/AmazonECR/latest/userguide/getting-started-cli.html#cli-create-repository 了解如何创建镜像仓库。

class torchx.schedulers.aws_batch_scheduler.AWSBatchScheduler(session_name: str, client: Optional[Any] = None, log_client: Optional[Any] = None, docker_client: Optional[DockerClient] = None)

基础： DockerWorkspaceMixin, Scheduler[AWSBatchOpts]

AWSBatchScheduler 是一个连接到 AWS Batch 的 TorchX 调度接口。

$ pip install torchx[kubernetes]
$ torchx run --scheduler aws_batch --scheduler_args queue=torchx utils.echo --image alpine:latest --msg hello
aws_batch://torchx_user/1234
$ torchx status aws_batch://torchx_user/1234
...

身份验证通过环境使用boto3凭证进行加载。

配置选项

    usage:
        queue=QUEUE,[user=USER],[privileged=PRIVILEGED],[share_id=SHARE_ID],[priority=PRIORITY],[image_repo=IMAGE_REPO]

    required arguments:
        queue=QUEUE (str)
            queue to schedule job in

    optional arguments:
        user=USER (str, ubuntu)
            The username to tag the job with. `getpass.getuser()` if not specified.
        privileged=PRIVILEGED (bool, False)
            If true runs the container with elevated permissions. Equivalent to running with `docker run --privileged`.
        share_id=SHARE_ID (str, None)
            The share identifier for the job. This must be set if and only if the job queue has a scheduling policy.
        priority=PRIORITY (int, 0)
            The scheduling priority for the job within the context of share_id. Higher number (between 0 and 9999) means higher priority. This will only take effect if the job queue has a scheduling policy.
        image_repo=IMAGE_REPO (str, None)
            (remote jobs) the image repository to use when pushing patched images, must have push access. Ex: example.com/your/container

挂载

此类支持绑定挂载主机目录、EFS 卷和主机设备。

• 绑定挂载： type=bind,src=<host path>,dst=<container path>[,readonly]

• 磁盘卷大小： type=volume,src=<efs id>,dst=<container path>[,readonly]

• devices: type=device,src=/dev/infiniband/uverbs0,[dst=<container path>][,perm=rwm]

参见 torchx.specs.parse_mounts() 以获取更多信息。

对于其他文件系统，如FSx，您可以将它们挂载到主机并将其绑定到您的工作：https://aws.amazon.com/premiumsupport/knowledge-center/batch-fsx-lustre-file-system-mount/

对于弹性织物适配器 (EFA)，您需要使用设备挂载将其挂载到容器中：https://docs.aws.amazon.com/batch/latest/userguide/efa.html

兼容性

功能	调度程序支持
获取日志	✔️
分布式作业	✔️
取消任务	✔️
描述工作	部分支持。AWSBatchScheduler 将返回作业和副本状态，但不提供完整的原始 AppSpec。
工作区 / 补丁修复	✔️
挂载	✔️
弹性	❌

describe(app_id: str) → Optional[DescribeAppResponse]

描述指定的应用程序。

Returns:

应用程序定义描述或 None 如果应用程序不存在。

list() → List[ListAppResponse]

对于调度器上发布的应用，此 API 返回一个 ListAppResponse 对象列表，每个对象包含应用 ID 及其状态。 注意：此 API 处于原型阶段，可能会发生变化。

log_iter(app_id: str, role_name: str, k: int = 0, regex: Optional[str] = None, since: Optional[datetime] = None, until: Optional[datetime] = None, should_tail: bool = False, streams: Optional[Stream] = None) → Iterable[str]

返回一个迭代器，用于访问日志行的 k``th replica of the ``role。 该迭代器在读取完所有符合条件的日志行后结束。

如果调度程序支持基于时间指针获取日志行，则since，until字段会被遵循，否则会被忽略。不指定since和until相当于获取所有可用的日志行。如果until为空，则迭代器的行为就像tail -f一样，跟随日志输出直到作业达到终端状态。

日志的确切定义取决于调度程序的具体设置。有些调度程序可能会将标准错误或标准输出视为日志，而其他调度程序则可能从日志文件中读取日志。

行为和假设：

1. 如果在不存在的应用程序上调用此方法，会产生未定义行为。 调用者应在调用此方法之前使用exists(app_id)检查应用程序是否存在。

2. 不是有状态的，用相同的参数调用此方法两次 会返回一个新的迭代器。之前的迭代 进度会丢失。

3. 不一定始终支持日志追尾功能。并非所有调度器都支持实时日志迭代（例如，在应用程序运行时追尾日志）。有关迭代器的行为，请参阅特定调度器的文档。

3.1 If the scheduler supports log-tailing, it should be controlled

by should_tail parameter.

4. 不保证日志保留。调用此方法时，底层调度程序可能已经清除了该应用程序的日志记录。如果是这样的话，此方法将引发任意异常。

5. 如果 should_tail 为 True，该方法仅在可访问的日志行已完全耗尽且应用程序达到最终状态时抛出 StopIteration 异常。 例如，如果应用程序卡住且没有生成任何日志行，则迭代器会阻塞直到应用程序最终被终止（通过超时或手动操作），此时它会抛出一个 StopIteration 异常。

   如果 should_tail 是 False，该方法在没有更多日志时抛出 StopIteration。

6. 不一定由所有调度器支持。

7. 某些调度器可能通过支持 __getitem__ （例如，iter[50] 寻找第 50 行日志）来支持行光标。

8. Whitespace is preserved, each new line should include \n. To

   支持交互式进度条，返回的行不需要包含 \n，但应该在不换行的情况下打印，以正确处理 \r 回车符。

Parameters:

流 – 要选择的IO输出流。 选项之一：combined、stdout、stderr。 如果所选流不被调度程序支持，它将抛出一个ValueError异常。

Returns:

一个指定角色副本的日志行上的 Iterator

Raises:

NotImplementedError – 如果调度器不支持日志迭代

schedule(dryrun_info: AppDryRunInfo[BatchJob]) → str

与 submit 相同，但接受一个 AppDryRunInfo。 建议实现者实现此方法而不是直接实现 submit，因为 submit 可以通过以下方式简单地实现：

dryrun_info = self.submit_dryrun(app, cfg)
return schedule(dryrun_info)

class torchx.schedulers.aws_batch_scheduler.BatchJob(name: str, queue: str, share_id: Union[str, NoneType], job_def: Dict[str, object], images_to_push: Dict[str, Tuple[str, str]])

参考

torchx.schedulers.aws_batch_scheduler.create_scheduler(session_name: str, **kwargs: object) → AWSBatchScheduler