dask_jobqueue.OARCluster
dask_jobqueue.OARCluster¶
- 类 dask_jobqueue.OARCluster(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)¶
在 OAR 集群上启动 Dask
- 参数
- queuestr
每个工作进程作业的目标队列。传递给 #OAR -q 选项。
- projectstr
与每个工作进程作业关联的项目。传递给 #OAR –project 选项。
- 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 or Bool
如果您使用 TLS/SSL，则为 dask.distributed 安全对象。如果为 True，将自动创建临时自签名凭据。
- scheduler_optionsdict
用于向 Dask Scheduler 传递额外参数。例如，使用
scheduler_options={'dashboard_address': ':12435'}指定 Web 控制面板应使用的端口，或使用scheduler_options={'host': 'your-host'}指定 Dask 调度器应运行的主机。参阅distributed.Scheduler获取更多详细信息。- scheduler_clstype
更改使用的 Dask Scheduler 类。默认为 Dask 的
distributed.Scheduler。- shared_temp_directorystr
调度器和工作进程之间的共享目录（例如用于临时安全证书），如果未设置，则默认为当前工作目录。
- resource_specstr
请求资源并指定作业位置。传递给 #OAR -l 选项。
- walltimestr
每个工作进程作业的墙钟时间。
- job_extralist
已弃用：请改用
job_extra_directives。此参数将在未来版本中移除。- job_extra_directiveslist
其他 OAR 选项列表，例如 -t besteffort。每个选项将添加 #OAR 前缀。
- memory_per_core_property_namestr
您的 OAR 集群的每核心内存属性名称（通常是 memcore 或 mem_core）。现有属性可以通过执行 oarnodes 命令列出。请注意，您的集群上可能不存在每核心内存属性。如果是这种情况，您可以将 memory_per_core_property_name=”not_applicable” 以避免收到警告。如果您将 memory_per_core_property_name 保持为其默认值，您将收到警告。如果 memory_per_core_property_name 是 “not_applicable” 或设置为其默认值，分配的节点可能没有足够的内存来匹配 memory 参数，并且 Dask 工作进程内存管理可能无法正常工作。
示例
>>> from dask_jobqueue import OARCluster >>> cluster = OARCluster(queue='regular') >>> 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