Resume and Caching

What you’ll learn¶

Use list_runs to check pipeline state in a delta root
Use resume to continue a pipeline from where it left off
Understand step-level caching for duplicate work
Configure CachePolicy options to control cache behavior

Every completed step in Artisan is persisted to delta storage. This enables two capabilities: resume (reload state and continue from the last step) and caching (detect duplicate work and skip re-execution). This tutorial walks through both.

Prerequisites: Your First Pipeline, Exploring Results.
Estimated time: 15 minutes
GPU required: No.

from __future__ import annotations

from artisan.operations.curator import Filter
from artisan.operations.examples import (
    DataGenerator,
    DataTransformer,
    MetricCalculator,
)
from artisan.orchestration import Backend, PipelineManager
from artisan.schemas.enums import CachePolicy
from artisan.utils import tutorial_setup
from artisan.visualization import build_macro_graph, inspect_pipeline

env = tutorial_setup("resume_and_caching")

Build the initial pipeline¶

Start with a 4-step pipeline that generates data, transforms it, computes metrics, and filters by a threshold. After it completes you will resume it and add more steps.

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

# Step 0: Generate 5 source datasets
pipeline.run(
    operation=DataGenerator,
    name="generate",
    params={"count": 5, "seed": 42},
    backend=Backend.LOCAL,
)

# Step 1: Transform each dataset
pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("generate", "datasets")},
    backend=Backend.LOCAL,
)

# Step 2: Compute metrics
pipeline.run(
    operation=MetricCalculator,
    name="score",
    inputs={"dataset": output("transform", "dataset")},
    backend=Backend.LOCAL,
)

# Step 3: Filter — keep datasets whose median > 0.3
pipeline.run(
    operation=Filter,
    name="filter",
    inputs={"passthrough": output("transform", "dataset")},
    params={
        "criteria": [
            {"metric": "distribution.median", "operator": "gt", "value": 0.3},
        ]
    },
    backend=Backend.LOCAL,
)

pipeline.finalize()
inspect_pipeline(env.delta_root)

List runs¶

Before resuming, you can check what runs exist in a delta root. PipelineManager.list_runs() returns a DataFrame with one row per pipeline run.

PipelineManager.list_runs(env.delta_root)

The DataFrame columns are:

Column	Meaning
`pipeline_run_id`	Unique identifier for the pipeline run
`step_count`	Number of completed steps in the run
`last_status`	Status of the most recent step
`started_at`	Timestamp when the run began
`ended_at`	Timestamp when the last step completed

Resume a pipeline¶

PipelineManager.resume() loads completed step state from the delta root and returns a PipelineManager positioned after the last completed step. New steps append to the existing run. If you have multiple runs in the same delta root, pass pipeline_run_id to select a specific one; otherwise the most recent run is used.

pipeline = PipelineManager.resume(
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)

# Add a 5th step: transform the filtered results
pipeline.run(
    operation=DataTransformer,
    name="refine",
    inputs={"dataset": pipeline[3].output("passthrough")},
    params={"seed": 99},
    backend=Backend.LOCAL,
)

pipeline.finalize()
inspect_pipeline(env.delta_root)

The output shows 5 steps now. Steps 0-3 were loaded from delta, step 4 was added fresh. pipeline[N] retrieves the StepResult for a previously completed step, which gives access to its .output() references for wiring into new downstream steps.

build_macro_graph(env.delta_root)

The macro graph shows the complete pipeline including both the original 4 steps and the resumed step 4.

Step-level caching¶

Artisan automatically detects when a step has already been executed with identical inputs and parameters. When this happens the step is served from cache instead of re-executing. The step spec ID -- a hash of the operation class, step number, parameters, and input references -- is what determines cache identity.

To demonstrate, create a second pipeline on the same delta root with overlapping steps.

# Re-open the same tutorial environment without cleaning
env = tutorial_setup("resume_and_caching", clean=False)

pipeline2 = PipelineManager.create(
    name="caching_demo",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
)
output = pipeline2.output

# These steps are identical to the first pipeline — they will be cached
pipeline2.run(
    operation=DataGenerator,
    name="generate",
    params={"count": 5, "seed": 42},
    backend=Backend.LOCAL,
)
pipeline2.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("generate", "datasets")},
    backend=Backend.LOCAL,
)
pipeline2.run(
    operation=MetricCalculator,
    name="score",
    inputs={"dataset": output("transform", "dataset")},
    backend=Backend.LOCAL,
)

# This step has different criteria — it will NOT be cached
pipeline2.run(
    operation=Filter,
    name="filter",
    inputs={"passthrough": output("transform", "dataset")},
    params={
        "criteria": [
            {"metric": "distribution.median", "operator": "gt", "value": 0.5},
        ]
    },
    backend=Backend.LOCAL,
)

pipeline2.finalize()
inspect_pipeline(env.delta_root, pipeline_run_id=pipeline2.config.pipeline_run_id)

In the log output you will see “CACHED” messages for steps 0-2 because they have identical operation, step number, params, and inputs. Step 3 has a different filter threshold (0.5 vs 0.3), so it executes fresh.

CachePolicy options¶

The cache policy controls how strict the cache hit requirements are. Artisan provides two policies:

Policy	Behavior	Use case
`ALL_SUCCEEDED` (default)	Cache hit only when the previous execution had zero failures	Production pipelines where partial results are unacceptable
`STEP_COMPLETED`	Cache hit for any completed step, even with some failed executions	Development / exploration where partial progress is acceptable

# Set at pipeline level (applies to all steps)
pipeline = PipelineManager.create(
    name="example",
    delta_root=env.delta_root,
    staging_root=env.staging_root,
    working_root=env.working_root,
    cache_policy=CachePolicy.STEP_COMPLETED,
)
print("Created pipeline with cache_policy=CachePolicy.STEP_COMPLETED")

The cache policy is set at pipeline creation time and applies to all steps in that run. Infrastructure errors (dispatch failures, commit errors) always block caching regardless of policy.

Summary¶

Concept	API	Purpose
List runs	`PipelineManager.list_runs(delta_root)`	Check what pipeline runs exist
Resume	`PipelineManager.resume(delta_root, staging_root)`	Continue a pipeline from where it left off
Step result access	`pipeline[N]`	Get output references for resumed steps
Caching	Automatic	Identical steps are served from cache
Cache policy	`CachePolicy.ALL_SUCCEEDED` / `STEP_COMPLETED`	Control cache hit requirements

Key takeaway: Resume and caching are built into the delta-based storage model. Every completed step is persisted to delta, enabling both resume (reload state and continue) and caching (detect duplicate work by step spec ID).

Next steps¶

Interactive Filter -- Explore metrics and set filter thresholds interactively
Timing Analysis -- Profile step and execution performance
Provenance Graphs -- Visualize pipeline topology and artifact lineage
Execution Flow -- How caching works under the hood (step-level and execution-level)
Storage and Delta Lake -- Why delta persistence enables resume and caching