dask_jobqueue.SLURMCluster

dask_jobqueue.SLURMCluster¶

class dask_jobqueue.SLURMCluster(n_workers=0, job_cls: typing.Optional[dask_jobqueue.core.Job] = None, loop=None, security=None, shared_temp_directory=None, silence_logs='error', name=None, asynchronous=False, dashboard_address=None, host=None, scheduler_options=None, scheduler_cls=<class 'distributed.scheduler.Scheduler'>, interface=None, protocol=None, config_name=None, **job_kwargs)¶

在 SLURM 集群上启动 Dask

参数

queuestr: 每个 worker 作业的目标队列。传递给 #SBATCH -p 选项。
projectstr: 已弃用：请改用 account。此参数将在未来版本中移除。
accountstr: 与每个 worker 作业关联的记账字符串。传递给 #SBATCH -A 选项。
coresint: 作业内所有 worker 线程运行的 CPU 核心总数。每个 worker 进程的线程数由公式 cores / processes 确定。作业队列系统默认将其用作每个作业的 CPU 数量。
memory: str: 作业内所有 worker 将使用的内存总量。作业队列系统默认将其用作每个作业的内存量。
processesint: 将作业分割成指定数量的进程。适用于 GIL 工作负载或具有许多核心的节点。默认情况下，process ~= sqrt(cores)，以便进程数与每个进程的线程数大致相同。
interfacestr: 网络接口，例如 ‘eth0’ 或 ‘ib0’。这将用于 Dask scheduler 和 Dask worker 接口。如果您需要为 Dask scheduler 使用不同的接口，可以通过 scheduler_options 参数传递：interface=your_worker_interface, scheduler_options={'interface': your_scheduler_interface}。
nannybool: 是否启动 nanny 进程
local_directorystr: Dask worker 用于文件溢出的本地目录。
death_timeoutfloat: 在关闭 worker 之前等待 scheduler 的秒数
extralist: 已弃用：请改用 worker_extra_args。此参数将在未来版本中移除。
worker_commandlist: 启动 worker 时运行的命令。默认为 “distributed.cli.dask_worker”
worker_extra_argslist: 传递给 dask-worker 的额外参数
env_extralist: 已弃用：请改用 job_script_prologue。此参数将在未来版本中移除。
job_script_prologuelist: 在启动 worker 之前添加到脚本中的其他命令。
job_script_epiloguelist: 添加到脚本中的命令，这些命令将在 worker 命令退出后运行。
header_skiplist: 已弃用：请改用 job_directives_skip。此参数将在未来版本中移除。
job_directives_skiplist: 在生成的作业脚本头部中要跳过的指令。包含指定字符串的指令行将被移除。通过 job_extra_directives 添加的指令不会受到影响。
log_directorystr: 用于作业调度器日志的目录。
shebangstr: 您的批处理提交脚本所需解释器的路径。
pythonstr: 用于启动 Dask worker 的 Python 可执行文件。默认为提交这些作业的 Python。
config_namestr: jobqueue.yaml 配置文件中要使用的部分。
namestr: Dask worker 的名称。这通常由 Cluster 设置。
n_workersint: 默认启动的 worker 数量。默认为 0。请参阅 scale 方法。
silence_logsstr: 如果 scheduler 在本地启动，此处发出的日志级别，如 “debug”、“info” 或 “error”
asynchronousbool: 是否使用 async/await 语法运行此集群对象。
securitySecurity 或 Bool: 如果您使用 TLS/SSL，这是一个 dask.distributed security 对象。如果为 True，将自动创建临时的自签名凭证。
scheduler_optionsdict: 用于向 Dask Scheduler 传递额外参数。例如，使用 scheduler_options={'dashboard_address': ':12435'} 指定 Web dashboard 应使用的端口，或使用 scheduler_options={'host': 'your-host'} 指定 Dask scheduler 应运行的主机。有关更多详细信息，请参阅 distributed.Scheduler。
scheduler_clstype: 更改所使用的 Dask Scheduler 类。默认为 Dask 的 distributed.Scheduler。
shared_temp_directorystr: scheduler 和 worker 之间的共享目录（例如用于临时安全证书），如果未设置，默认为当前工作目录。
walltimestr: 每个 worker 作业的 walltime（总运行时间上限）。
job_cpuint: 在 SLURM 中为每个作业请求的 CPU 数量。对于某些复杂的非 Python 代码，此选项可能有助于请求比 worker 线程总数更多的 CPU。如果为 None，默认为 cores。
job_memstr: 在 SLURM 中为每个作业请求的内存量，如果为 None，默认为 memory。
job_extralist: 已弃用：请改用 job_extra_directives。此参数将在未来版本中移除。
job_extra_directiveslist: 其他 Slurm 选项的列表，例如 -j oe。每个选项都将加上 #SBATCH 前缀。

示例

>>> from dask_jobqueue import SLURMCluster
>>> cluster = SLURMCluster(
...     queue='regular',
...     account="myaccount",
...     cores=24,
...     memory="500 GB"
... )
>>> cluster.scale(jobs=10)  # ask for 10 jobs

>>> from dask.distributed import Client
>>> client = Client(cluster)

这也适用于自适应集群。它会根据负载自动启动和终止 worker。

>>> cluster.adapt(maximum_jobs=20)

__init__(n_workers=0, job_cls: typing.Optional[dask_jobqueue.core.Job] = None, loop=None, security=None, shared_temp_directory=None, silence_logs='error', name=None, asynchronous=False, dashboard_address=None, host=None, scheduler_options=None, scheduler_cls=<class 'distributed.scheduler.Scheduler'>, interface=None, protocol=None, config_name=None, **job_kwargs)¶

方法

`__init__`([n_workers, job_cls, loop, ...])
`adapt`(*args[, minimum_jobs, maximum_jobs])	根据 scheduler 活动自动调整 Dask 集群的规模。
`close`([timeout])
`from_name`(name)	通过名称创建一个此类实例来表示现有集群。
`get_client`()	返回集群的客户端
`get_logs`([cluster, scheduler, workers])	返回集群、scheduler 和 worker 的日志
`job_script`()
`logs`(args, *kwargs)
`new_worker_spec`()	返回下一个 worker 的名称和规格
`scale`([n, jobs, memory, cores])	将集群扩展到指定的配置。
`scale_down`(workers)
`scale_up`([n, memory, cores])	将集群扩展到 n 个 worker
`sync`(func, *args[, asynchronous, ...])	根据调用上下文同步或异步地调用 func 函数并传入 args 参数
`wait_for_workers`(n_workers[, timeout])	阻塞调用，等待 n 个 worker 后继续。

属性

`asynchronous`	是否正在事件循环中运行？
`called_from_running_loop`
`dashboard_link`
`job_header`
`job_name`
`loop`
`name`
`observed`
`plan`
`requested`
`scheduler_address`

dask_jobqueue.SGECluster

概述

Dask-jobqueue 0.9.0+10.g7cbd4b8 文档

dask_jobqueue.SLURMCluster

dask_jobqueue.SLURMCluster¶