dask_jobqueue.PBSCluster
dask_jobqueue.PBSCluster¶
- class dask_jobqueue.PBSCluster(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)¶
在 PBS 集群上启动 Dask
- 参数
- queuestr
每个工作者作业的目标队列。传递给 #PBS -q 选项。
- projectstr
已弃用：请改用
account。此参数将在未来版本中移除。- accountstr
与每个工作者作业关联的记账字符串。传递给 #PBS -A 选项。
- coresint
一个作业中所有工作者线程运行的 CPU 核心总数。每个工作者进程的线程数使用公式
cores / processes确定。作业队列系统默认将其用作每个作业的 CPU 数量。- memory: str
一个作业中所有工作者使用的内存总量。作业队列系统默认将其用作每个作业的内存量。
- processesint
将作业分成这么多进程。适用于 GIL 工作负载或具有许多核心的节点。默认情况下，
process ~= sqrt(cores)，以便进程数和每个进程的线程数大致相同。- interfacestr
网络接口，如 ‘eth0’ 或 ‘ib0’。这将用于 Dask 调度器和 Dask 工作者接口。如果您需要为 Dask 调度器使用不同的接口，可以通过
scheduler_options参数传递：interface=your_worker_interface, scheduler_options={'interface': your_scheduler_interface}。- nannybool
是否启动 nanny 进程
- local_directorystr
用于文件溢出的 Dask 工作者本地目录。
- death_timeoutfloat
在关闭工作者之前等待调度器的秒数
- extralist
已弃用：请改用
worker_extra_args。此参数将在未来版本中移除。- worker_commandlist
启动工作者时运行的命令。默认为 “distributed.cli.dask_worker”
- worker_extra_argslist
传递给 dask-worker 的额外参数
- env_extralist
已弃用：请改用
job_script_prologue。此参数将在未来版本中移除。- job_script_prologuelist
在启动工作者之前添加到脚本的其他命令。
- job_script_epiloguelist
工作者命令退出后添加到脚本中运行的命令。
- header_skiplist
已弃用：请改用
job_directives_skip。此参数将在未来版本中移除。- job_directives_skiplist
在生成的作业脚本头部中要跳过的指令。包含指定字符串的指令行将被移除。通过 job_extra_directives 添加的指令不受影响。
- log_directorystr
用于作业调度器日志的目录。
- shebangstr
批处理提交脚本所需解释器的路径。
- pythonstr
用于启动 Dask 工作者的 Python 可执行文件。默认为提交这些作业的 Python。
- config_namestr
jobqueue.yaml 配置文件中要使用的部分。
- namestr
Dask 工作者的名称。这通常由 Cluster 设置。
- n_workersint
默认启动的工作者数量。默认为 0。参见 scale 方法。
- silence_logsstr
如果调度器在本地启动，在此处发出的日志级别，如“debug”、“info”或“error”。
- asynchronousbool
是否使用 async/await 语法运行此集群对象。
- securitySecurity 或 Bool
如果您使用 TLS/SSL，则为 dask.distributed 安全对象。如果为 True，将自动创建临时的自签名凭证。
- scheduler_optionsdict
用于向 Dask 调度器传递额外参数。例如，使用
scheduler_options={'dashboard_address': ':12435'}指定 Web 仪表板应使用的端口，或使用scheduler_options={'host': 'your-host'}指定 Dask 调度器应运行的主机。更多详细信息请参见distributed.Scheduler。- scheduler_clstype
更改所使用的 Dask 调度器类。默认为 Dask 的
distributed.Scheduler。- shared_temp_directorystr
调度器和工作者之间的共享目录（例如用于临时安全证书），如果未设置，则默认为当前工作目录。
- resource_specstr
请求资源并指定作业放置。传递给 #PBS -l 选项。
- walltimestr
每个工作者作业的墙钟时间。
- job_extralist
已弃用：请改用
job_extra_directives。此参数将在未来版本中移除。- job_extra_directiveslist
其他 PBS 选项列表。每个选项前面都会加上 #PBS 前缀。
示例
>>> from dask_jobqueue import PBSCluster >>> cluster = PBSCluster(queue='regular', account="myaccountingstring", cores=24, ... memory="500 GB") >>> cluster.scale(jobs=10) # ask for 10 jobs
>>> from dask.distributed import Client >>> client = Client(cluster)
这也适用于自适应集群。它根据负载自动启动和终止工作者。
>>> 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])根据调度器活动自动伸缩 Dask 集群。
close([timeout])from_name(name)通过名称创建此类的实例来表示现有集群。
get_client()返回集群的客户端
get_logs([cluster, scheduler, workers])返回集群、调度器和工作者的日志
job_script()logs(*args, **kwargs)new_worker_spec()返回下一个工作者的名称和规格
scale([n, jobs, memory, cores])将集群伸缩到指定配置。
scale_down(workers)scale_up([n, memory, cores])将集群伸缩到 n 个工作者
sync(func, *args[, asynchronous, ...])根据调用上下文同步或异步调用 func 并传递 args
wait_for_workers(n_workers[, timeout])阻塞调用，等待 n 个工作者准备好后继续
属性
asynchronous我们是否在事件循环中运行？
called_from_running_loopdashboard_linkjob_headerjob_nameloopnameobservedplanrequestedscheduler_address