1. 引言aiida-flux-scheduler 是一个面向 AiiDA 工作流引擎的调度器插件用于将 AiiDA 的计算任务提交到 Flux 调度系统上运行。Flux 是 Lawrence Livermore 国家实验室开发的新一代高性能计算调度框架具有低延迟、高扩展性和灵活的资源管理能力。通过 aiida-flux-scheduler研究人员可以在 AiiDA 中无缝使用 Flux 作为计算后端实现工作流与调度器的深度集成。本文将从功能特性、安装配置、核心语法与参数、9 个实际应用案例以及常见错误与注意事项等方面系统性地介绍 aiida-flux-scheduler 的使用方法。2. 功能概述aiida-flux-scheduler 的核心功能包括以下几个方面作业提交将 AiiDA 计算节点转换为 Flux 作业并提交到集群。状态轮询定期查询 Flux 作业状态同步到 AiiDA 数据库。资源分配支持节点数、CPU 数、GPU 数等资源的灵活配置。队列管理支持指定 Flux 队列Queue和银行账户Bank等调度参数。作业取消支持取消正在运行或排队中的作业。输出捕获自动捕获作业的标准输出和标准错误存入 AiiDA 节点。环境变量传递支持向作业传递自定义环境变量。多集群支持通过配置多个计算资源可对接不同的 Flux 集群。3. 安装方法3.1 环境要求安装 aiida-flux-scheduler 前需要满足以下环境要求Python 3.8 及以上版本AiiDA 2.0 及以上版本可访问的 Flux 调度系统本地或远程pip 包管理器3.2 通过 pip 安装推荐使用 pip 直接安装最新稳定版本pip install aiida-flux-scheduler如果需要安装特定版本可以指定版本号pip install aiida-flux-scheduler0.3.13.3 从源码安装对于需要二次开发或使用最新特性的用户可以从 GitHub 仓库克隆并安装git clone https://github.com/aiidateam/aiida-flux-scheduler.git cd aiida-flux-scheduler pip install -e .3.4 验证安装安装完成后可以通过以下命令验证插件是否正确注册verdi plugin list aiida.schedulers在输出列表中应能看到flux调度器条目。4. 核心语法与参数4.1 计算资源配置使用 aiida-flux-scheduler 前需要在 AiiDA 中配置计算资源Computer。以下是一个典型的配置示例verdi computer create -L flux-cluster -H login01.example.org -T flux -S core -w /home/user/aiida-work flux-cluster其中关键参数说明-T flux指定传输方式为 SSH 直连。-S core指定调度器类型为 Flux。-w指定工作目录。4.2 作业提交参数在 AiiDA 计算节点中通过CalcJob的options字典配置 Flux 调度参数。常用参数如下options { resources: { num_machines: 2, num_mpiprocs_per_machine: 32, num_cores_per_machine: 64, }, queue_name: batch, account: project123, max_wallclock_seconds: 3600, custom_scheduler_commands: #FLUX: gpus4, environment_variables: { OMP_NUM_THREADS: 16, MY_VAR: hello, }, job_name: my-flux-job, }参数说明resources资源字典包含节点数、每节点 MPI 进程数、每节点核心数。queue_nameFlux 队列名称。account计费账户。max_wallclock_seconds最大运行时长秒。custom_scheduler_commands自定义 Flux 指令以#FLUX:开头。environment_variables传递给作业的环境变量字典。job_name作业名称。4.3 提交作业的 Python 代码模板以下是一个完整的作业提交示例from aiida.engine import run from aiida.orm import Code, Dict from aiida.plugins import DataFactory 加载计算代码 code Code.get(flux-testflux-cluster) 构造输入参数 parameters Dict(dict{message: Hello from Flux!}) 提交计算 builder code.get_builder() builder.parameters parameters builder.metadata.options.resources { num_machines: 1, num_mpiprocs_per_machine: 1, } builder.metadata.options.max_wallclock_seconds 300 builder.metadata.options.queue_name batch result run(builder) print(计算完成输出, result)5. 9 个实际应用案例5.1 案例一Hello World 基础提交最简单的入门案例验证 Flux 调度器与 AiiDA 的连通性from aiida.engine import run from aiida.orm import Code, Dict from aiida.plugins import DataFactory code Code.get(echoflux-cluster) parameters Dict(dict{text: Hello, Flux!}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.max_wallclock_seconds 60 result run(builder) print(输出内容, result[stdout].get_content())5.2 案例二多节点 MPI 并行计算使用 Flux 调度器提交跨节点 MPI 作业适用于大规模并行仿真from aiida.engine import run from aiida.orm import Code, Dict code Code.get(mpi_programflux-cluster) parameters Dict(dict{iterations: 1000, grid_size: 256}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources { num_machines: 4, num_mpiprocs_per_machine: 32, } builder.metadata.options.max_wallclock_seconds 7200 builder.metadata.options.queue_name parallel result run(builder) print(MPI 作业完成)5.3 案例三GPU 作业提交通过自定义调度指令申请 GPU 资源适用于深度学习训练任务from aiida.engine import run from aiida.orm import Code, Dict code Code.get(pytorch_trainflux-cluster) parameters Dict(dict{epochs: 50, batch_size: 128}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources { num_machines: 1, num_cores_per_machine: 16, } builder.metadata.options.custom_scheduler_commands #FLUX: gpus4 builder.metadata.options.max_wallclock_seconds 10800 result run(builder) print(GPU 训练完成)5.4 案例四参数扫描批量作业结合 AiiDA 的 workchain 机制批量提交多个参数组合的作业from aiida.engine import WorkChain, run from aiida.orm import Code, Dict, List class ParameterScan(WorkChain): classmethod def define(cls, spec): super().define(spec) spec.input(code, valid_typeCode) spec.input(parameters_list, valid_typeList) spec.outline(cls.run_all) def run_all(self): for params in self.inputs.parameters_list: builder self.inputs.code.get_builder() builder.parameters Dict(dictparams) builder.metadata.options.resources {num_machines: 1} builder.metadata.options.max_wallclock_seconds 600 self.submit(builder) parameters_list List(list[ {temperature: 300, pressure: 1.0}, {temperature: 350, pressure: 1.5}, {temperature: 400, pressure: 2.0}, ]) result run(ParameterScan, codeCode.get(simflux-cluster), parameters_listparameters_list) print(参数扫描完成)5.5 案例五作业依赖链通过 AiiDA 的数据流依赖自动构建作业依赖链前一个作业的输出作为后一个作业的输入from aiida.engine import run from aiida.orm import Code, Dict code Code.get(pipeline_stepflux-cluster) 第一步 builder1 code.get_builder() builder1.parameters Dict(dict{stage: preprocess, input_file: raw.dat}) builder1.metadata.options.resources {num_machines: 1} result1 run(builder1) 第二步依赖第一步输出 builder2 code.get_builder() builder2.parameters Dict(dict{stage: analyze, input_file: result1[output_file].value}) builder2.metadata.options.resources {num_machines: 2} result2 run(builder2) print(依赖链执行完成)5.6 案例六环境变量传递向作业传递自定义环境变量适用于需要配置运行时环境的场景from aiida.engine import run from aiida.orm import Code, Dict code Code.get(env_testflux-cluster) parameters Dict(dict{task: print_env}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.environment_variables { MY_CUSTOM_VAR: custom_value, OMP_NUM_THREADS: 8, LD_LIBRARY_PATH: /opt/custom/lib, } result run(builder) print(环境变量测试完成)5.7 案例七长时间运行作业配置长时间运行的作业并设置合理的最大运行时长from aiida.engine import run from aiida.orm import Code, Dict code Code.get(long_simflux-cluster) parameters Dict(dict{simulation_time: 100000, output_interval: 1000}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources { num_machines: 8, num_mpiprocs_per_machine: 64, } builder.metadata.options.max_wallclock_seconds 172800 # 48 小时 builder.metadata.options.queue_name long result run(builder) print(长时间作业完成)5.8 案例八作业取消与重试在 workchain 中实现作业失败后的自动取消与重试逻辑from aiida.engine import WorkChain, run, if_ from aiida.orm import Code, Dict class RetryWorkChain(WorkChain): classmethod def define(cls, spec): super().define(spec) spec.input(code, valid_typeCode) spec.input(parameters, valid_typeDict) spec.outline( cls.submit_job, if_(cls.is_failed)(cls.retry_job), cls.finalize, ) def submit_job(self): builder self.inputs.code.get_builder() builder.parameters self.inputs.parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.max_wallclock_seconds 300 self.ctx.job self.submit(builder) def is_failed(self): return self.ctx.job.is_failed def retry_job(self): self.report(作业失败正在重试...) builder self.inputs.code.get_builder() builder.parameters self.inputs.parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.max_wallclock_seconds 600 self.ctx.job self.submit(builder) def finalize(self): self.report(作业最终完成) result run(RetryWorkChain, codeCode.get(flakyflux-cluster), parametersDict(dict{x: 1})) print(重试工作链完成)5.9 案例九多集群混合调度配置多个 Flux 集群根据作业类型动态选择计算资源from aiida.engine import WorkChain, run from aiida.orm import Code, Dict class MultiClusterWorkChain(WorkChain): classmethod def define(cls, spec): super().define(spec) spec.input(cpu_code, valid_typeCode) spec.input(gpu_code, valid_typeCode) spec.input(parameters, valid_typeDict) spec.outline(cls.run_on_cpu, cls.run_on_gpu) def run_on_cpu(self): builder self.inputs.cpu_code.get_builder() builder.parameters self.inputs.parameters builder.metadata.options.resources {num_machines: 2} builder.metadata.options.queue_name cpu-queue self.ctx.cpu_result self.submit(builder) def run_on_gpu(self): builder self.inputs.gpu_code.get_builder() builder.parameters self.inputs.parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.custom_scheduler_commands #FLUX: gpus2 builder.metadata.options.queue_name gpu-queue self.ctx.gpu_result self.submit(builder) result run( MultiClusterWorkChain, cpu_codeCode.get(cpu_programflux-cpu-cluster), gpu_codeCode.get(gpu_programflux-gpu-cluster), parametersDict(dict{data: input.dat}), ) print(多集群混合调度完成)6. 常见错误与使用注意事项6.1 常见错误错误类型错误信息示例解决方法调度器未注册Unknown scheduler type: flux重新安装插件并执行verdi plugin list aiida.schedulers确认注册连接失败SSH connection timed out检查网络连通性、SSH 密钥配置和主机名是否正确资源参数错误Invalid resources specification检查num_machines、num_mpiprocs_per_machine等参数是否为正整数队列不存在Queue batch not found使用flux queue list查看可用队列修改《动手学PyTorch建模与应用:从深度学习到大模型》是一本从零基础上手深度学习和大模型的PyTorch实战指南。全书共11章前6章涵盖深度学习基础包括张量运算、神经网络原理、数据预处理及卷积神经网络等后5章进阶探讨图像、文本、音频建模技术并结合Transformer架构解析大语言模型的开发实践。书中通过房价预测、图像分类等案例讲解模型构建方法每章附有动手练习题帮助读者巩固实战能力。内容兼顾数学原理与工程实现适配PyTorch框架最新技术发展趋势。
1. 引言aiida-flux-scheduler 是一个面向 AiiDA 工作流引擎的调度器插件用于将 AiiDA 的计算任务提交到 Flux 调度系统上运行。Flux 是 Lawrence Livermore 国家实验室开发的新一代高性能计算调度框架具有低延迟、高扩展性和灵活的资源管理能力。通过 aiida-flux-scheduler研究人员可以在 AiiDA 中无缝使用 Flux 作为计算后端实现工作流与调度器的深度集成。本文将从功能特性、安装配置、核心语法与参数、9 个实际应用案例以及常见错误与注意事项等方面系统性地介绍 aiida-flux-scheduler 的使用方法。2. 功能概述aiida-flux-scheduler 的核心功能包括以下几个方面作业提交将 AiiDA 计算节点转换为 Flux 作业并提交到集群。状态轮询定期查询 Flux 作业状态同步到 AiiDA 数据库。资源分配支持节点数、CPU 数、GPU 数等资源的灵活配置。队列管理支持指定 Flux 队列Queue和银行账户Bank等调度参数。作业取消支持取消正在运行或排队中的作业。输出捕获自动捕获作业的标准输出和标准错误存入 AiiDA 节点。环境变量传递支持向作业传递自定义环境变量。多集群支持通过配置多个计算资源可对接不同的 Flux 集群。3. 安装方法3.1 环境要求安装 aiida-flux-scheduler 前需要满足以下环境要求Python 3.8 及以上版本AiiDA 2.0 及以上版本可访问的 Flux 调度系统本地或远程pip 包管理器3.2 通过 pip 安装推荐使用 pip 直接安装最新稳定版本pip install aiida-flux-scheduler如果需要安装特定版本可以指定版本号pip install aiida-flux-scheduler0.3.13.3 从源码安装对于需要二次开发或使用最新特性的用户可以从 GitHub 仓库克隆并安装git clone https://github.com/aiidateam/aiida-flux-scheduler.git cd aiida-flux-scheduler pip install -e .3.4 验证安装安装完成后可以通过以下命令验证插件是否正确注册verdi plugin list aiida.schedulers在输出列表中应能看到flux调度器条目。4. 核心语法与参数4.1 计算资源配置使用 aiida-flux-scheduler 前需要在 AiiDA 中配置计算资源Computer。以下是一个典型的配置示例verdi computer create -L flux-cluster -H login01.example.org -T flux -S core -w /home/user/aiida-work flux-cluster其中关键参数说明-T flux指定传输方式为 SSH 直连。-S core指定调度器类型为 Flux。-w指定工作目录。4.2 作业提交参数在 AiiDA 计算节点中通过CalcJob的options字典配置 Flux 调度参数。常用参数如下options { resources: { num_machines: 2, num_mpiprocs_per_machine: 32, num_cores_per_machine: 64, }, queue_name: batch, account: project123, max_wallclock_seconds: 3600, custom_scheduler_commands: #FLUX: gpus4, environment_variables: { OMP_NUM_THREADS: 16, MY_VAR: hello, }, job_name: my-flux-job, }参数说明resources资源字典包含节点数、每节点 MPI 进程数、每节点核心数。queue_nameFlux 队列名称。account计费账户。max_wallclock_seconds最大运行时长秒。custom_scheduler_commands自定义 Flux 指令以#FLUX:开头。environment_variables传递给作业的环境变量字典。job_name作业名称。4.3 提交作业的 Python 代码模板以下是一个完整的作业提交示例from aiida.engine import run from aiida.orm import Code, Dict from aiida.plugins import DataFactory 加载计算代码 code Code.get(flux-testflux-cluster) 构造输入参数 parameters Dict(dict{message: Hello from Flux!}) 提交计算 builder code.get_builder() builder.parameters parameters builder.metadata.options.resources { num_machines: 1, num_mpiprocs_per_machine: 1, } builder.metadata.options.max_wallclock_seconds 300 builder.metadata.options.queue_name batch result run(builder) print(计算完成输出, result)5. 9 个实际应用案例5.1 案例一Hello World 基础提交最简单的入门案例验证 Flux 调度器与 AiiDA 的连通性from aiida.engine import run from aiida.orm import Code, Dict from aiida.plugins import DataFactory code Code.get(echoflux-cluster) parameters Dict(dict{text: Hello, Flux!}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.max_wallclock_seconds 60 result run(builder) print(输出内容, result[stdout].get_content())5.2 案例二多节点 MPI 并行计算使用 Flux 调度器提交跨节点 MPI 作业适用于大规模并行仿真from aiida.engine import run from aiida.orm import Code, Dict code Code.get(mpi_programflux-cluster) parameters Dict(dict{iterations: 1000, grid_size: 256}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources { num_machines: 4, num_mpiprocs_per_machine: 32, } builder.metadata.options.max_wallclock_seconds 7200 builder.metadata.options.queue_name parallel result run(builder) print(MPI 作业完成)5.3 案例三GPU 作业提交通过自定义调度指令申请 GPU 资源适用于深度学习训练任务from aiida.engine import run from aiida.orm import Code, Dict code Code.get(pytorch_trainflux-cluster) parameters Dict(dict{epochs: 50, batch_size: 128}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources { num_machines: 1, num_cores_per_machine: 16, } builder.metadata.options.custom_scheduler_commands #FLUX: gpus4 builder.metadata.options.max_wallclock_seconds 10800 result run(builder) print(GPU 训练完成)5.4 案例四参数扫描批量作业结合 AiiDA 的 workchain 机制批量提交多个参数组合的作业from aiida.engine import WorkChain, run from aiida.orm import Code, Dict, List class ParameterScan(WorkChain): classmethod def define(cls, spec): super().define(spec) spec.input(code, valid_typeCode) spec.input(parameters_list, valid_typeList) spec.outline(cls.run_all) def run_all(self): for params in self.inputs.parameters_list: builder self.inputs.code.get_builder() builder.parameters Dict(dictparams) builder.metadata.options.resources {num_machines: 1} builder.metadata.options.max_wallclock_seconds 600 self.submit(builder) parameters_list List(list[ {temperature: 300, pressure: 1.0}, {temperature: 350, pressure: 1.5}, {temperature: 400, pressure: 2.0}, ]) result run(ParameterScan, codeCode.get(simflux-cluster), parameters_listparameters_list) print(参数扫描完成)5.5 案例五作业依赖链通过 AiiDA 的数据流依赖自动构建作业依赖链前一个作业的输出作为后一个作业的输入from aiida.engine import run from aiida.orm import Code, Dict code Code.get(pipeline_stepflux-cluster) 第一步 builder1 code.get_builder() builder1.parameters Dict(dict{stage: preprocess, input_file: raw.dat}) builder1.metadata.options.resources {num_machines: 1} result1 run(builder1) 第二步依赖第一步输出 builder2 code.get_builder() builder2.parameters Dict(dict{stage: analyze, input_file: result1[output_file].value}) builder2.metadata.options.resources {num_machines: 2} result2 run(builder2) print(依赖链执行完成)5.6 案例六环境变量传递向作业传递自定义环境变量适用于需要配置运行时环境的场景from aiida.engine import run from aiida.orm import Code, Dict code Code.get(env_testflux-cluster) parameters Dict(dict{task: print_env}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.environment_variables { MY_CUSTOM_VAR: custom_value, OMP_NUM_THREADS: 8, LD_LIBRARY_PATH: /opt/custom/lib, } result run(builder) print(环境变量测试完成)5.7 案例七长时间运行作业配置长时间运行的作业并设置合理的最大运行时长from aiida.engine import run from aiida.orm import Code, Dict code Code.get(long_simflux-cluster) parameters Dict(dict{simulation_time: 100000, output_interval: 1000}) builder code.get_builder() builder.parameters parameters builder.metadata.options.resources { num_machines: 8, num_mpiprocs_per_machine: 64, } builder.metadata.options.max_wallclock_seconds 172800 # 48 小时 builder.metadata.options.queue_name long result run(builder) print(长时间作业完成)5.8 案例八作业取消与重试在 workchain 中实现作业失败后的自动取消与重试逻辑from aiida.engine import WorkChain, run, if_ from aiida.orm import Code, Dict class RetryWorkChain(WorkChain): classmethod def define(cls, spec): super().define(spec) spec.input(code, valid_typeCode) spec.input(parameters, valid_typeDict) spec.outline( cls.submit_job, if_(cls.is_failed)(cls.retry_job), cls.finalize, ) def submit_job(self): builder self.inputs.code.get_builder() builder.parameters self.inputs.parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.max_wallclock_seconds 300 self.ctx.job self.submit(builder) def is_failed(self): return self.ctx.job.is_failed def retry_job(self): self.report(作业失败正在重试...) builder self.inputs.code.get_builder() builder.parameters self.inputs.parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.max_wallclock_seconds 600 self.ctx.job self.submit(builder) def finalize(self): self.report(作业最终完成) result run(RetryWorkChain, codeCode.get(flakyflux-cluster), parametersDict(dict{x: 1})) print(重试工作链完成)5.9 案例九多集群混合调度配置多个 Flux 集群根据作业类型动态选择计算资源from aiida.engine import WorkChain, run from aiida.orm import Code, Dict class MultiClusterWorkChain(WorkChain): classmethod def define(cls, spec): super().define(spec) spec.input(cpu_code, valid_typeCode) spec.input(gpu_code, valid_typeCode) spec.input(parameters, valid_typeDict) spec.outline(cls.run_on_cpu, cls.run_on_gpu) def run_on_cpu(self): builder self.inputs.cpu_code.get_builder() builder.parameters self.inputs.parameters builder.metadata.options.resources {num_machines: 2} builder.metadata.options.queue_name cpu-queue self.ctx.cpu_result self.submit(builder) def run_on_gpu(self): builder self.inputs.gpu_code.get_builder() builder.parameters self.inputs.parameters builder.metadata.options.resources {num_machines: 1} builder.metadata.options.custom_scheduler_commands #FLUX: gpus2 builder.metadata.options.queue_name gpu-queue self.ctx.gpu_result self.submit(builder) result run( MultiClusterWorkChain, cpu_codeCode.get(cpu_programflux-cpu-cluster), gpu_codeCode.get(gpu_programflux-gpu-cluster), parametersDict(dict{data: input.dat}), ) print(多集群混合调度完成)6. 常见错误与使用注意事项6.1 常见错误错误类型错误信息示例解决方法调度器未注册Unknown scheduler type: flux重新安装插件并执行verdi plugin list aiida.schedulers确认注册连接失败SSH connection timed out检查网络连通性、SSH 密钥配置和主机名是否正确资源参数错误Invalid resources specification检查num_machines、num_mpiprocs_per_machine等参数是否为正整数队列不存在Queue batch not found使用flux queue list查看可用队列修改《动手学PyTorch建模与应用:从深度学习到大模型》是一本从零基础上手深度学习和大模型的PyTorch实战指南。全书共11章前6章涵盖深度学习基础包括张量运算、神经网络原理、数据预处理及卷积神经网络等后5章进阶探讨图像、文本、音频建模技术并结合Transformer架构解析大语言模型的开发实践。书中通过房价预测、图像分类等案例讲解模型构建方法每章附有动手练习题帮助读者巩固实战能力。内容兼顾数学原理与工程实现适配PyTorch框架最新技术发展趋势。
