Configure Execution

How to control where operations run, what resources they get, and how work is batched — from local development through production SLURM.

Prerequisites: Operations Model, Building a Pipeline

Key types: Backend, ResourceConfig, ExecutionConfig, ToolSpec, Environments, CachePolicy, FailurePolicy

Minimal working example¶

A pipeline running one step locally and one on SLURM with GPU resources:

from artisan.orchestration import Backend, PipelineManager
from myops import PreprocessOp, InferenceOp

pipeline = PipelineManager.create(
    name="example",
    delta_root="runs/delta",
    staging_root="runs/staging",
)

pipeline.run(operation=PreprocessOp, name="preprocess", params={"count": 100})

pipeline.run(
    operation=InferenceOp,
    name="inference",
    inputs={"dataset": pipeline.output("preprocess", "dataset")},
    backend=Backend.SLURM,
    resources={"gpus": 1, "memory_gb": 32, "extra": {"partition": "gpu"}},
    execution={"artifacts_per_unit": 1},
)

The rest of this guide breaks down each option.

Choose a compute backend¶

Every step runs on a compute backend. Set it per step or as a pipeline-wide default:

from artisan.orchestration import Backend

# Pipeline-wide default
pipeline = PipelineManager.create(..., backend=Backend.SLURM)

# Step-level override
pipeline.run(operation=MyOp, inputs=..., backend=Backend.LOCAL)

For SLURM_INTRA, you must be inside an existing SLURM allocation (salloc or sbatch). Work is distributed via srun with no queue wait:

pipeline.run(
    operation=MyOp,
    inputs=...,
    backend=Backend.SLURM_INTRA,
    resources={"gpus": 1, "cpus": 4, "memory_gb": 16},
)

For LOCAL, you can cap the number of concurrent workers per step:

pipeline.run(operation=MyOp, inputs=..., execution={"max_workers": 8})

The default process pool size is 4.

Configure resources¶

Pass a resources dict to override resource allocation for a step:

pipeline.run(
    operation=MyOp,
    inputs=...,
    backend=Backend.SLURM,
    resources={
        "gpus": 1,
        "memory_gb": 32,
        "time_limit": "04:00:00",
        "cpus": 4,
        "extra": {"partition": "gpu"},
    },
)

ResourceConfig fields¶

ResourceConfig is portable across backends — each backend translates these fields to its native format. Use extra for backend-specific settings like SLURM partition or account.

Step-level resources merge with operation defaults — you only need to specify the fields you want to override.

Control batching¶

Batching determines how many artifacts each worker processes. This is the main lever for tuning throughput.

pipeline.run(
    operation=MyOp,
    inputs=...,
    execution={"artifacts_per_unit": 10},
)

With 100 input artifacts and artifacts_per_unit=10, the framework creates 10 execution units, each processing a batch of 10 artifacts.

Two-level batching¶

Batching happens at two levels:

100 artifacts
    │
    │  artifacts_per_unit = 10
    ▼
10 execution units (logical work packages)
    │
    │  units_per_worker = 2
    ▼
5 SLURM jobs (each runs 2 units sequentially)

Level 1 — artifacts_per_unit: How many artifacts each execution unit processes. Set this based on your operation’s workload: 1 for GPU inference (one artifact per job), 50–100 for fast metrics calculations.

Level 2 — units_per_worker: How many execution units a single SLURM job runs sequentially. Use this to amortize job startup overhead without changing your operation’s batch logic.

ExecutionConfig fields¶

Set operation-level defaults¶

Operations can declare their own default resources and execution config so you don’t repeat the same overrides at every step:

from artisan.operations.base import OperationDefinition
from artisan.schemas.operation_config.resource_config import ResourceConfig
from artisan.schemas.execution.execution_config import ExecutionConfig

class GpuInference(OperationDefinition):
    name = "gpu_inference"

    resources: ResourceConfig = ResourceConfig(
        gpus=1,
        memory_gb=32,
        time_limit="02:00:00",
        extra={"partition": "gpu"},
    )

    execution: ExecutionConfig = ExecutionConfig(
        artifacts_per_unit=1,
        estimated_seconds=600.0,
    )

    # ... lifecycle methods ...

Step-level overrides merge on top of these defaults. For example, to give a specific step more memory without changing other settings:

pipeline.run(operation=GpuInference, inputs=..., resources={"memory_gb": 64})
# gpus, time_limit, extra keep their operation defaults

Override precedence¶

Pipeline defaults (PipelineManager.create)
    └── Operation defaults (class fields)
            └── Step overrides (pipeline.run kwargs)   ← wins

Configure external tools and environments¶

Operations that wrap external tools declare two things: a ToolSpec (the binary/script to invoke) and an Environments configuration (the runtime that wraps the command):

from pathlib import Path
from artisan.operations.base import OperationDefinition
from artisan.schemas.operation_config.tool_spec import ToolSpec
from artisan.schemas.operation_config.environments import Environments
from artisan.schemas.operation_config.environment_spec import ApptainerEnvironmentSpec

class ToolAOp(OperationDefinition):
    name = "tool_a"

    tool: ToolSpec = ToolSpec(
        executable=Path("run_tool_a.sh"),
        interpreter="bash",
    )

    environments: Environments = Environments(
        active="apptainer",
        apptainer=ApptainerEnvironmentSpec(
            image=Path("/tools/tool_a.sif"),
            gpu=True,
            binds=[
                (Path("/data/weights"), Path("/weights")),
                (Path("/scratch"), Path("/scratch")),
            ],
        ),
    )

    # ... lifecycle methods ...

Override tool or environment settings at the step level:

pipeline.run(
    operation=ToolAOp,
    inputs=...,
    tool={"executable": "run_tool_a_v2.sh"},
    environment={"apptainer": {"image": "/tools/tool_a_v2.sif"}},
)

When you pass a dict for environment, fields are deep-merged with the operation’s existing environment config. This means partial overrides work without discarding other fields. To switch the active environment without changing any spec fields, pass a string instead:

pipeline.run(operation=ToolAOp, inputs=..., environment="local")

The binds field takes a list of (host_path, container_path) tuples — not colon-delimited strings.

ToolSpec fields¶

Environment spec types¶

All specs share a base EnvironmentSpec with an env dict for extra environment variables.

Set failure policy¶

Control what happens when some artifacts fail within a step:

from artisan.schemas.enums import FailurePolicy

# Pipeline-wide default
pipeline = PipelineManager.create(..., failure_policy=FailurePolicy.CONTINUE)

# Step-level override
pipeline.run(operation=MyOp, inputs=..., failure_policy=FailurePolicy.FAIL_FAST)

CONTINUE is the default because in large runs (thousands of artifacts), a single malformed input should not discard thousands of successful results. Failures are always recorded for diagnosis.

Set cache policy¶

Cache policy controls when a previously completed step qualifies as a cache hit on re-run (e.g., when resuming a pipeline):

from artisan.schemas.enums import CachePolicy

pipeline = PipelineManager.create(..., cache_policy=CachePolicy.STEP_COMPLETED)

Both policies block caching when infrastructure errors (dispatch or commit failures) occurred. The difference is whether partial-failure steps count as hits.

Use STEP_COMPLETED when you want to skip re-running a step that mostly succeeded, even if a few artifacts failed.

Use non-blocking execution¶

pipeline.run() blocks until the step completes. For steps that can overlap (e.g., independent branches), use pipeline.submit() to dispatch without waiting:

future = pipeline.submit(
    operation=BranchAOp,
    inputs={"data": pipeline.output("preprocess", "data")},
    backend=Backend.SLURM,
)

# Submit another step concurrently
pipeline.submit(
    operation=BranchBOp,
    inputs={"data": pipeline.output("preprocess", "data")},
    backend=Backend.SLURM,
)

# Downstream steps that depend on a submitted step automatically wait
pipeline.run(
    operation=MergeOp,
    inputs={"a": pipeline.output("branch_a", "result"),
            "b": pipeline.output("branch_b", "result")},
)

submit() returns a StepFuture. The orchestrator tracks dependencies and blocks downstream steps until their inputs are ready.

Common patterns¶

Development: inspectable sandboxes¶

During development, you can make the working directory visible and persistent:

pipeline = PipelineManager.create(
    ...,
    working_root="runs/working",
    preserve_working=True,
)

This writes sandboxes to runs/working/ instead of $TMPDIR and keeps them after execution completes, so you can inspect input materialization and output files.

For production, omit working_root — the default uses $TMPDIR (typically node-local SSD on SLURM clusters), which avoids shared filesystem contention.

Debugging: preserve staging files¶

pipeline = PipelineManager.create(..., preserve_staging=True)

Keeps the raw Parquet files workers produce before commit. Useful for diagnosing staging or commit issues.

Recovering from crashes¶

By default, PipelineManager.create commits leftover staging files from prior crashed runs at pipeline initialization (recover_staging=True). To disable this:

pipeline = PipelineManager.create(..., recover_staging=False)

Naming steps¶

By default, each step is named after the operation. Provide a custom name to disambiguate when the same operation appears multiple times:

pipeline.run(operation=ScoreOp, name="score_round1", inputs=...)
pipeline.run(operation=ScoreOp, name="score_round2", inputs=...)

# Reference by name
pipeline.output("score_round1", "scores")

Tuning SLURM throughput¶

For operations with fast per-artifact execution (< 1 second), increase artifacts_per_unit to reduce job overhead:

pipeline.run(
    operation=FastMetrics,
    inputs=...,
    backend=Backend.SLURM,
    execution={"artifacts_per_unit": 100, "units_per_worker": 5},
)

For GPU operations, keep artifacts_per_unit=1 and let SLURM handle parallelism via job arrays.

Custom SLURM parameters¶

Use extra for backend-specific parameters not covered by ResourceConfig:

pipeline.run(
    operation=MyOp,
    inputs=...,
    resources={
        "extra": {
            "partition": "gpu",
            "constraint": "a100",
            "account": "my_allocation",
            "exclude": "node[001-003]",
        }
    },
)

Disabling Delta Lake compaction¶

Each run() call compacts Delta Lake tables after commit. To skip compaction (useful when running many small steps in sequence):

pipeline.run(operation=MyOp, inputs=..., compact=False)

Common pitfalls¶

Verify¶

Confirm your configuration works by running a small test:

step = pipeline.run(operation=MyOp, inputs=..., backend=Backend.LOCAL)
assert step.success
print(f"Processed {step.succeeded_count} artifacts")

Then switch to Backend.SLURM for production. Check SLURM job logs if failures occur — the job name format is s{step_number}_{operation_name}.

Cross-references¶

• Execution Flow — dispatch, execute, commit lifecycle

• SLURM Execution Tutorial — interactive SLURM walkthrough

• Writing Creator Operations — declaring operation-level defaults