dask_jobqueue.SGECluster

dask_jobqueue.SGECluster¶

类 dask_jobqueue.SGECluster(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)¶

在 SGE 集群上启动 Dask

注意

如果需要指定 RAM 大小，必须同时指定 memory 和 resource_spec。 resource_spec 的确切语法由您的 GridEngine 系统管理员定义。请求的 memory 大小应与 resource_spec 匹配，以便 Dask 的内存管理系统能够准确执行。

参数

queuestr: 每个 worker 作业的目标队列。传递给 #$ -q 选项。
projectstr: 与每个 worker 作业关联的项目。传递给 #$ -P 选项。
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: 关闭 workers 前等待 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 or 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 之间的共享目录（例如用于临时安全证书）。如果未设置，默认为当前工作目录。
resource_specstr: 请求资源并指定作业位置。传递给 #$ -l 选项。
walltimestr: 每个 worker 作业的壁钟时间（walltime）。
job_extralist: 已废弃：请改用 job_extra_directives。此参数将在未来的版本中移除。
job_extra_directiveslist: 其他 SGE 选项的列表，例如 -w e。每个选项都会被加上 #$ 前缀。

示例

>>> from dask_jobqueue import SGECluster
>>> cluster = SGECluster(
...     queue='regular',
...     project="myproj",
...     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`()	返回集群的 client
`get_logs`([cluster, scheduler, workers])	返回集群、scheduler 和 workers 的日志
`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.PBSCluster

dask_jobqueue.SLURMCluster

Dask-jobqueue 0.9.0+10.g7cbd4b8 文档

dask_jobqueue.SGECluster

dask_jobqueue.SGECluster¶