Hey Nate! How can I help today? If you’re testing the assistant, I’m ready to draft examples or walk through anything—especially around container-based deployments, ECS, or EC2.

Ahh got it — thanks for clarifying. Short answer: there isn’t a built-in Slurm worker in Prefect 3.x. You’ve got two solid patterns depending on what you mean by “workers that run on a Slurm cluster”: 1) Easiest: run standard Prefect “process” workers inside long-lived Slurm jobs - When to use: you want a pool of persistent workers scheduled by Slurm, each running flows as local subprocesses within its Slurm allocation. - How it looks: - Create a process work pool:

prefect work-pool create slurm-process-pool --type process

- Submit Slurm jobs that just start a worker (use arrays to scale N workers):

#!/bin/bash
    #SBATCH --job-name=prefect-worker
    #SBATCH --partition=compute
    #SBATCH --time=08:00:00
    #SBATCH --cpus-per-task=8
    #SBATCH --mem=32G
    #SBATCH --output=/path/to/logs/prefect-%A_%a.out

    module load python/3.11  # if needed
    source /path/to/venv/bin/activate

    export PREFECT_API_URL="<https://api.prefect.cloud/api/accounts/.../workspaces/...>"
    export PREFECT_API_KEY="***"

    # Optionally pin to a specific work queue and concurrency limit
    prefect worker start --pool slurm-process-pool --limit 1

- Scale by submitting an array (e.g.,

--array=1-10

) so you get 10 workers. - Pros: No custom code; leverages Slurm to schedule worker capacity; very stable. - Tradeoff: Each flow run does not get its own Slurm job; runs share the allocation of the worker that picks them up. 2) Custom Slurm worker that submits each flow run as an sbatch job - When to use: you want one Slurm job per flow run with Slurm-native isolation (partition/account/time/mem) and lifecycle tracking. - How it looks (outline): - Implement a

BaseWorker

subclass that: - Builds an

sbatch

script using the flow run’s

command

and your Slurm params - Calls

sbatch

to submit - Monitors status via

squeue/sacct/scontrol

- Returns a

BaseWorkerResult

with the Slurm job ID and exit code - Minimal skeleton: ``` from pydantic import Field from prefect.workers.base import BaseWorker, BaseJobConfiguration, BaseVariables, BaseWorkerResult import anyio, asyncio class SlurmJobConfiguration(BaseJobConfiguration): partition: str = Field(default="compute") time_limit: str = Field(default="010000") cpus_per_task: int = Field(default=1) mem: str = Field(default="4G") class SlurmVariables(BaseVariables): partition: str = Field(default="compute") time_limit: str = Field(default="010000") cpus_per_task: int = Field(default=1) mem: str = Field(default="4G") class SlurmWorkerResult(BaseWorkerResult): slurm_job_id: str | None = None class SlurmWorker(BaseWorker): type = "slurm" job_configuration = SlurmJobConfiguration job_configuration_variables = SlurmVariables async def run(self, flow_run, configuration, task_status=anyio.TASK_STATUS_IGNORED): sbatch_script = self._build_sbatch_script(configuration, command=configuration.command) job_id, submit_out = await self._submit_sbatch(sbatch_script) if task_status: task_status.started(job_id) exit_code = await self._wait_for_completion(job_id) return SlurmWorkerResult(status_code=exit_code, identifier=job_id, slurm_job_id=job_id) def _build_sbatch_script(self, cfg, command: str) -> str: lines = [ "#!/bin/bash", f"#SBATCH --job-name={cfg.name or 'prefect-flow'}", f"#SBATCH --partition={cfg.partition}", f"#SBATCH --time={cfg.time_limit}", f"#SBATCH --cpus-per-task={cfg.cpus_per_task}", f"#SBATCH --mem={cfg.mem}", ] for k, v in (cfg.env or {}).items(): lines.append(f"export {k}={v}") lines.append(command) return "\n".join(lines) + "\n"

async def _submit_sbatch(self, script: str): import tempfile with tempfile.NamedTemporaryFile(mode="w", suffix=".sh", delete=False) as f: f.write(script) path = f.name proc = await asyncio.create_subprocess_exec("sbatch", path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) out, err = await proc.communicate() job_id = out.decode().strip().split()[-1] # "Submitted batch job 12345" return job_id, out.decode() async def _wait_for_completion(self, job_id: str) -> int: while True: proc = await asyncio.create_subprocess_exec("sacct", "-j", job_id, "--format=State,ExitCode", "--parsable2", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) out, _ = await proc.communicate() text = out.decode().strip() if "COMPLETED" in text or "FAILED" in text or "CANCELLED" in text or "TIMEOUT" in text: # Parse ExitCode=0:0 style for line in text.splitlines(): if line and ":" in line: parts = line.split("|") if len(parts) >= 2 and parts[1]: code = parts[1].split(":")[0] return int(code) return 1 await anyio.sleep(5)

- Register via entry points so the CLI discovers it, then:

prefect work-pool create slurm-pool --type slurm prefect worker start --pool slurm-pool ``` - Pros: True “one flow run = one Slurm job”; native resource controls per run. - Tradeoff: You maintain a small plugin. Helpful docs: - Custom worker guide: Developing a custom worker (Prefect 3) - Work pools and base job templates: Manage work pools - Worker concepts: Workers concept Quick check of CLI: - Create a pool:

prefect work-pool create NAME --type <type>

(verified with --help) - Start a worker:

prefect worker start --pool NAME

(type comes from the pool) Which model are you aiming for? - A) Workers themselves scheduled by Slurm (simplest; use process pool + sbatch) - B) Each flow run is a Slurm job (requires a small custom worker) If you pick one, I can sketch a minimal repo (pyproject + worker class + template) or an sbatch array script tailored to your cluster config.