Running on SLURM

What you’ll learn¶

Switch a pipeline from local to SLURM execution with a single parameter
Mix local and SLURM steps in the same pipeline
Override SLURM resources (partition, memory, GPUs) per step
Control batching to tune job count vs. work per job
Debug SLURM runs with preserve_working and preserve_staging

Prerequisites: First Pipeline, Run vs Submit, SLURM cluster access. Estimated time: 10 minutes GPU required: No (uses CPU partition for demonstration).

from __future__ import annotations

from artisan.operations.examples import (
    DataGenerator,
    DataTransformer,
    MetricCalculator,
)
from artisan.orchestration import Backend, PipelineManager
from artisan.utils import tutorial_setup
from artisan.visualization import inspect_metrics, inspect_pipeline

env = tutorial_setup("slurm_execution")
DELTA_ROOT = env.delta_root

The one-parameter switch¶

Every pipeline.run() and pipeline.submit() call accepts a backend parameter. Set it to Backend.SLURM and the framework submits that step as a SLURM job array. Everything else — operation class, inputs, params, output wiring — stays identical.

# Local (default)
step = pipeline.run(MyOperation, inputs=..., backend=Backend.LOCAL)

# SLURM — same operation, same inputs, different backend
step = pipeline.run(MyOperation, inputs=..., backend=Backend.SLURM)

This means you can develop and debug locally, then move to the cluster by changing one argument per step.

Building a mixed local/SLURM pipeline¶

A common pattern is to run lightweight steps locally and send compute-intensive steps to the cluster:

Step	Operation	Backend	Why
0	`DataGenerator`	LOCAL	Fast — creates small test datasets
1	`DataTransformer`	SLURM	Compute-intensive transformation
2	`MetricCalculator`	SLURM	Compute-intensive metric calculation

The pipeline wires outputs to inputs the same way regardless of backend. Let’s build it.

pipeline = PipelineManager.create(
    name="slurm_tutorial",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)
output = pipeline.output

Step 0: Generate data (local)¶

DataGenerator is fast — no reason to wait for SLURM scheduling overhead.

step0 = pipeline.run(
    operation=DataGenerator,
    name="generate",
    params={"count": 4, "seed": 42},
    backend=Backend.LOCAL,
)
print(f"Generated {step0.succeeded_count} datasets locally")

Step 1: Transform data (SLURM)¶

The only change: backend=Backend.SLURM. Each execution unit becomes a SLURM job. You’ll see the job in sjobs or squeue named s1_data_transformer.

step1 = pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 0.5, "variants": 2, "seed": 100},
    backend=Backend.SLURM,
)
print(f"Transformed {step1.succeeded_count} datasets on SLURM")

Step 2: Compute metrics (SLURM)¶

Output wiring works the same across backends — output("transform", "dataset") resolves to the artifacts produced by the SLURM step.

step2 = pipeline.run(
    operation=MetricCalculator,
    name="metrics",
    inputs={"dataset": output("transform", "dataset")},
    backend=Backend.SLURM,
)
print(f"Computed metrics for {step2.succeeded_count} datasets on SLURM")

summary = pipeline.finalize()
print(
    f"Pipeline complete: {summary['total_steps']} steps, success={summary['overall_success']}"
)

Inspect results¶

The same inspection tools work regardless of backend. Results land in the same Delta Lake tables whether steps ran locally or on SLURM.

inspect_pipeline(DELTA_ROOT)

inspect_metrics(DELTA_ROOT, step_number=2)

Overriding SLURM resources¶

Every operation has default resources (memory, time limit, CPUs, etc.). Override any field per step with the resources parameter. The SLURM backend maps these portable fields to SLURM directives automatically.

Common fields: cpus, memory_gb, gpus, time_limit, and extra (a dict for arbitrary SLURM flags like {"partition": "gpu"}).

The following examples use a fresh pipeline to demonstrate overrides.

pipeline = PipelineManager.create(
    name="slurm_overrides",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)
output = pipeline.output

# Generate source data for the override examples
step0 = pipeline.run(
    operation=DataGenerator,
    name="generate",
    params={"count": 4, "seed": 42},
    backend=Backend.LOCAL,
)
print(f"Generated {step0.succeeded_count} datasets for override examples")

# Check an operation's default resources
defaults = DataTransformer.model_fields["resources"].default
print("DataTransformer defaults:")
print(f"  cpus:       {defaults.cpus}")
print(f"  memory_gb:  {defaults.memory_gb}")
print(f"  gpus:       {defaults.gpus}")
print(f"  time_limit: {defaults.time_limit}")
print(f"  extra:      {defaults.extra}")

Override specific fields at run() time — unspecified fields keep their defaults:

# Request more memory and CPUs for a compute-heavy step
step = pipeline.run(
    operation=DataTransformer,
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 0.5, "variants": 1, "seed": 200},
    backend=Backend.SLURM,
    resources={
        "memory_gb": 12,
        "cpus": 4,
        "time_limit": "04:00:00",
        "extra": {"partition": "cpu"},
    },
)
print(f"Transformed {step.succeeded_count} datasets with custom resources")

Controlling batching¶

Batching controls how artifacts are grouped into SLURM jobs. Two levels:

artifacts_per_unit — How many input artifacts go into one execution unit. With 100 artifacts and artifacts_per_unit=10, the framework creates 10 execution units. This means 10 artifacts are submitted to each process.
units_per_worker — How many execution units a single SLURM job processes sequentially. With 10 units and units_per_worker=2, the framework submits 5 SLURM jobs.

100 artifacts
  ÷ artifacts_per_unit=10  →  10 execution units
  ÷ units_per_worker=2    →   5 SLURM jobs

When to increase artifacts_per_unit: Your operation processes artifacts quickly and SLURM scheduling overhead dominates.

When to increase units_per_worker: You want fewer total SLURM jobs (e.g., to stay within array size limits or reduce scheduler load).

# Check an operation's default batching
exec_defaults = MetricCalculator.model_fields["execution"].default
print("MetricCalculator defaults:")
print(f"  artifacts_per_unit: {exec_defaults.artifacts_per_unit}")
print(f"  units_per_worker:   {exec_defaults.units_per_worker}")
print(f"  job_name:           {exec_defaults.job_name}")

# Override batching: 5 artifacts per unit, 2 units per SLURM job
step = pipeline.run(
    operation=MetricCalculator,
    inputs={"dataset": output("generate", "datasets")},
    backend=Backend.SLURM,
    execution={"artifacts_per_unit": 5, "units_per_worker": 2},
)
print(f"Computed metrics for {step.succeeded_count} datasets with custom batching")

Non-blocking SLURM with `submit()`¶

pipeline.submit() works with SLURM the same way it works locally — it returns a StepFuture immediately while the SLURM jobs run in the background. This is useful for overlapping independent SLURM steps.

pipeline = PipelineManager.create(
    name="slurm_submit_demo",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)
output = pipeline.output

pipeline.run(
    operation=DataGenerator,
    name="generate",
    params={"count": 4, "seed": 42},
    backend=Backend.LOCAL,
)

# Submit two independent SLURM steps
future_a = pipeline.submit(
    operation=DataTransformer,
    name="transform_a",
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 0.5, "variants": 1, "seed": 100},
    backend=Backend.SLURM,
)

future_b = pipeline.submit(
    operation=DataTransformer,
    name="transform_b",
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 2.0, "variants": 1, "seed": 200},
    backend=Backend.SLURM,
)

# Both are now running on the cluster concurrently
print(f"Step 1 status: {future_a.status}")
print(f"Step 2 status: {future_b.status}")

# Collect results
result_a = future_a.result()
result_b = future_b.result()
print(f"Step 1: {result_a.succeeded_count} artifacts")
print(f"Step 2: {result_b.succeeded_count} artifacts")

pipeline.finalize()

Debugging SLURM runs¶

Develop locally first¶

The fastest debugging strategy: run with Backend.LOCAL until your pipeline logic is correct, then switch to SLURM for production.

Monitor jobs with `squeue`¶

SLURM job names follow the pattern s{step}_{operation}, making them straightforward to filter:

$ squeue -u $USER
  JOBID  PARTITION  NAME                STATE   TIME
  12345  cpu        s1_data_transformer RUNNING 0:15
  12346  cpu        s2_metric_calculator PENDING 0:00

Preserve working directories¶

Pass preserve_working=True when creating the pipeline to keep worker sandbox directories after execution. This lets you inspect the files each worker received and produced.

# Enable debug flags at pipeline creation
debug_pipeline = PipelineManager.create(
    name="slurm_debug",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
    preserve_working=True,  # keep worker sandboxes after execution
    preserve_staging=True,  # keep staged Parquet files after commit
)

print(f"preserve_working: {debug_pipeline.config.preserve_working}")
print(f"preserve_staging: {debug_pipeline.config.preserve_staging}")

With these flags enabled:

preserve_working=True: The working_root directory retains each worker’s sandbox after execution. Browse it to see materialized input files, command outputs, and logs.
preserve_staging=True: The staging_root directory retains Parquet files after they’ve been committed to Delta Lake. Useful for verifying what was committed.

Both directories are cleaned up by default to save disk space.

Summary¶

One parameter switches between local and SLURM: backend
Operations, inputs, params, and output wiring are identical across backends
Override resources per step with resources={"memory_gb": ..., "cpus": ...}
Control batching with execution={"artifacts_per_unit": ..., "units_per_worker": ...}
Use submit() for non-blocking SLURM steps that run concurrently
Debug with preserve_working=True and preserve_staging=True
Develop locally first, switch to SLURM for production

Next steps¶

Running Inside a SLURM Allocation — Zero-latency dispatch via srun within an existing salloc session
Writing Your First Operation — Build a custom operation
Pipeline Patterns — Reusable pipeline topologies (branching, filtering, multi-input)
Execution Flow — How the framework dispatches and tracks work

Running on SLURM

What you’ll learn¶

The one-parameter switch¶

Building a mixed local/SLURM pipeline¶

Step 0: Generate data (local)¶

Step 1: Transform data (SLURM)¶

Step 2: Compute metrics (SLURM)¶

Inspect results¶

Overriding SLURM resources¶

Controlling batching¶

Non-blocking SLURM with submit()¶

Debugging SLURM runs¶

Develop locally first¶

Monitor jobs with squeue¶

Preserve working directories¶

Summary¶

Next steps¶

Non-blocking SLURM with `submit()`¶

Monitor jobs with `squeue`¶