Composites

What you’ll learn¶

Define reusable operation compositions with CompositeDefinition
Run composites collapsed (pipeline.run()) or expanded (pipeline.expand())
Control intermediate artifact handling (discard, persist, expose)
Nest composites inside composites

Prerequisites: Sources and Sequencing, Advanced Patterns
Estimated time: 15 minutes

When operations are tightly coupled — for example, transform → score where you always score immediately after transforming — defining them as a composite lets you treat them as a single reusable unit.

A CompositeDefinition declares its inputs, outputs, and internal wiring via a compose() method. The pipeline then decides how to execute it:

Mode	Method	Steps	I/O between internal ops	Use when
Collapsed	`pipeline.run(MyComposite, ...)`	1	In-memory	Operations always run together
Expanded	`pipeline.expand(MyComposite, ...)`	N	Delta Lake	You want independent caching/batching per op

from __future__ import annotations

from artisan.composites import (
    CompositeContext,
    CompositeDefinition,
)
from artisan.operations.examples import (
    DataGenerator,
    DataTransformer,
    MetricCalculator,
)
from artisan.orchestration import PipelineManager
from artisan.utils import tutorial_setup
from artisan.visualization import (
    build_macro_graph,
    build_micro_graph,
    inspect_pipeline,
)

Graph legend: See Sources and Sequencing for box/arrow key.

Defining a composite¶

A composite is a subclass of CompositeDefinition. It declares:

InputRole / OutputRole — StrEnum classes naming the composite’s external inputs and outputs
inputs / outputs — ClassVar dicts mapping roles to InputSpec / OutputSpec
Params — an optional Pydantic BaseModel for composite-level parameters
compose() — the wiring method that calls ctx.run() to execute internal operations

Here’s a composite that transforms data and then computes quality metrics:

from enum import StrEnum
from typing import ClassVar

from pydantic import BaseModel, Field

from artisan.schemas.specs.input_spec import InputSpec
from artisan.schemas.specs.output_spec import OutputSpec


class TransformAndScore(CompositeDefinition):
    """Transform data and compute quality metrics in one step."""

    name = "transform_and_score"
    description = "Transform data, then compute metrics."

    class InputRole(StrEnum):
        DATASET = "dataset"

    class OutputRole(StrEnum):
        METRICS = "metrics"

    inputs: ClassVar[dict[str, InputSpec]] = {
        InputRole.DATASET: InputSpec(artifact_type="data", required=True),
    }
    outputs: ClassVar[dict[str, OutputSpec]] = {
        OutputRole.METRICS: OutputSpec(
            artifact_type="metric", description="Quality metrics"
        ),
    }

    class Params(BaseModel):
        scale_factor: float = Field(
            default=2.0, description="Scale factor for transform"
        )

    params: Params = Params()

    def compose(self, ctx: CompositeContext) -> None:
        transformed = ctx.run(
            DataTransformer,
            inputs={"dataset": ctx.input("dataset")},
            params={
                "scale_factor": self.params.scale_factor,
                "variants": 1,
                "seed": 100,
            },
        )
        scored = ctx.run(
            MetricCalculator,
            inputs={"dataset": transformed.output("dataset")},
        )
        ctx.output("metrics", scored.output("metrics"))

The composite is now a reusable building block. It accepts datasets, runs two internal operations, and exposes only the final metrics. The compose() method uses CompositeContext to wire everything together — ctx.input() references declared inputs, ctx.run() executes operations, and ctx.output() maps results to declared outputs.

Baseline: separate `run()` calls¶

First, let’s build a three-step pipeline the traditional way — generate data, transform it, then score it. Each step reads from and writes to Delta Lake.

generate ──→ transform ──→ score
   (3 datasets)  (3 datasets)  (3 metrics)

This produces 3 steps and 3 Delta Lake round-trips.

env_baseline = tutorial_setup("baseline")

pipeline = PipelineManager.create(
    name="baseline",
    delta_root=env_baseline.delta_root,
    staging_root=env_baseline.staging_root,
    working_root=env_baseline.working_root,
)
output = pipeline.output

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

# Step 1: Transform each dataset
pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 2.0, "variants": 1, "seed": 100},
)

# Step 2: Score each transformed dataset
pipeline.run(
    operation=MetricCalculator,
    name="score",
    inputs={"dataset": output("transform", "dataset")},
)

result_baseline = pipeline.finalize()

inspect_pipeline(env_baseline.delta_root)

Three steps, three Delta Lake commits. The transform step wrote 3 intermediate datasets that were immediately read back by the score step. If these operations are always run together, that intermediate I/O is wasted work.

build_macro_graph(env_baseline.delta_root)

build_micro_graph(env_baseline.delta_root)

Collapsed mode: `pipeline.run()`¶

Passing a CompositeDefinition subclass to pipeline.run() executes it in collapsed mode — the entire composite runs as a single pipeline step. Internal artifacts are passed in-memory between operations within one worker, and only the declared outputs are committed to Delta Lake.

generate ──→ [ transform → score ]  ← single step
   (3 datasets)      (3 metrics)

This produces 2 steps instead of 3.

env_collapsed = tutorial_setup("collapsed")

pipeline = PipelineManager.create(
    name="collapsed",
    delta_root=env_collapsed.delta_root,
    staging_root=env_collapsed.staging_root,
    working_root=env_collapsed.working_root,
)
output = pipeline.output

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

# Step 1: Run the composite as a single step
pipeline.run(
    operation=TransformAndScore,
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 2.0},
)

result_collapsed = pipeline.finalize()

inspect_pipeline(env_collapsed.delta_root)

build_macro_graph(env_collapsed.delta_root)

build_micro_graph(env_collapsed.delta_root)

Two steps instead of three. The composite produced the same 3 metrics as the baseline, but the intermediate transformed datasets were never written to Delta Lake — they were passed directly in-memory from DataTransformer to MetricCalculator within each worker.

The macro graph shows the composite as a single node. From the pipeline’s perspective, transform_and_score is one step that accepts datasets and produces metrics.

Expanded mode: `pipeline.expand()`¶

pipeline.expand() dissolves the composite into separate pipeline steps. Each internal ctx.run() call becomes its own step with independent worker dispatch, batching, and caching. Step names are prefixed with the composite name.

generate ──→ transform_and_score.data_transformer ──→ transform_and_score.metric_calculator
   (3 datasets)            (3 datasets)                        (3 metrics)

This produces 3 steps, the same as the baseline, but from the same composite definition.

env_expanded = tutorial_setup("expanded")

pipeline = PipelineManager.create(
    name="expanded",
    delta_root=env_expanded.delta_root,
    staging_root=env_expanded.staging_root,
    working_root=env_expanded.working_root,
)
output = pipeline.output

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

# Expand the composite into separate pipeline steps
expanded = pipeline.expand(
    TransformAndScore,
    inputs={"dataset": output("generate", "datasets")},
)

result_expanded = pipeline.finalize()

inspect_pipeline(env_expanded.delta_root)

build_macro_graph(env_expanded.delta_root)

build_micro_graph(env_expanded.delta_root)

Three steps, with composite-prefixed names. Each internal operation got its own Delta Lake commit and can be cached independently. The same TransformAndScore definition drives both collapsed and expanded execution — the pipeline caller decides the mode.

Controlling intermediates¶

In collapsed mode, intermediate artifacts (outputs from internal operations that are not declared composite outputs) can be handled in three ways:

Mode	Intermediates in Delta	Use when
`"discard"` (default)	No	Production: minimize storage
`"persist"`	Yes (not visible to downstream steps)	Debugging: inspect intermediate results
`"expose"`	Yes (visible as step outputs)	Downstream steps need intermediate outputs

Pass intermediates="persist" to pipeline.run() to keep the intermediate artifacts in Delta Lake:

env_persist = tutorial_setup("persist")

pipeline = PipelineManager.create(
    name="persist",
    delta_root=env_persist.delta_root,
    staging_root=env_persist.staging_root,
    working_root=env_persist.working_root,
)
output = pipeline.output

# Step 0: Generate
pipeline.run(operation=DataGenerator, name="generate", params={"count": 2, "seed": 42})

# Step 1: Composite with intermediates="persist"
pipeline.run(
    operation=TransformAndScore,
    inputs={"dataset": output("generate", "datasets")},
    intermediates="persist",
)

result_persist = pipeline.finalize()

inspect_pipeline(env_persist.delta_root)

build_macro_graph(env_persist.delta_root)

build_micro_graph(env_persist.delta_root)

With persist, the composite committed both the intermediate transformed datasets and the final metrics to Delta Lake. The pipeline overview shows both artifact types in the composite step’s output. The micro provenance graph includes internal edges alongside the step-boundary edges, giving full visibility into the composite’s internal data flow.

Nesting composites¶

Composites can contain other composites. The inner composite executes recursively within the outer one’s compose() method. In collapsed mode, everything runs in-memory on a single worker. In expanded mode, each internal operation becomes its own step.

Here’s a composite that generates data and then applies TransformAndScore:

class GenerateAndScore(CompositeDefinition):
    """Generate data, then transform and score it."""

    name = "generate_and_score"
    description = "Full pipeline: generate, transform, score."

    class OutputRole(StrEnum):
        METRICS = "metrics"

    outputs: ClassVar[dict[str, OutputSpec]] = {
        OutputRole.METRICS: OutputSpec(
            artifact_type="metric", description="Quality metrics"
        ),
    }

    class Params(BaseModel):
        count: int = Field(default=2, description="Number of datasets to generate")
        scale_factor: float = Field(
            default=2.0, description="Scale factor for transform"
        )

    params: Params = Params()

    def compose(self, ctx: CompositeContext) -> None:
        # Generate data
        generated = ctx.run(
            DataGenerator,
            params={"count": self.params.count, "seed": 42},
        )
        # Nest TransformAndScore composite
        scored = ctx.run(
            TransformAndScore,
            inputs={"dataset": generated.output("datasets")},
            params={"scale_factor": self.params.scale_factor},
        )
        ctx.output("metrics", scored.output("metrics"))

env_nested = tutorial_setup("nested")

pipeline = PipelineManager.create(
    name="nested",
    delta_root=env_nested.delta_root,
    staging_root=env_nested.staging_root,
    working_root=env_nested.working_root,
)

# The entire pipeline is a single collapsed step
pipeline.run(operation=GenerateAndScore, params={"count": 2, "scale_factor": 1.5})

result_nested = pipeline.finalize()

inspect_pipeline(env_nested.delta_root)

build_macro_graph(env_nested.delta_root)

build_micro_graph(env_nested.delta_root)

The outer composite (GenerateAndScore) ran DataGenerator, then delegated to the inner composite (TransformAndScore), which ran DataTransformer and MetricCalculator — all as a single pipeline step. Nesting lets you build larger composites from smaller ones while keeping each piece independently testable.

Summary¶

This tutorial covered composable operations with CompositeDefinition:

Composite definition — subclass CompositeDefinition, declare InputRole/OutputRole enums, inputs/outputs specs, optional Params, and implement compose() using CompositeContext.
Collapsed mode — pipeline.run(MyComposite, ...) executes the composite as a single pipeline step with in-memory artifact passing.
Expanded mode — pipeline.expand(MyComposite, ...) dissolves the composite into separate pipeline steps, each with independent caching, batching, and worker dispatch.
Intermediates — intermediates="discard" (default) drops intermediate artifacts; "persist" keeps them in Delta Lake for debugging; "expose" makes them visible as step-boundary outputs.
Nesting — composites can contain other composites, executing recursively in either mode.

	Collapsed (`run`)	Expanded (`expand`)
Steps	1	N (one per internal op)
Internal I/O	In-memory	Delta Lake
Caching	Composite-level	Per-operation
Batching	Composite-level	Per-operation
Intermediates	Configurable	Always persisted

Key takeaway: Define your composition once as a CompositeDefinition, then let the pipeline caller choose collapsed or expanded execution based on the use case.

Next steps¶

Composites and Composition — conceptual deep dive into collapsed vs expanded execution, intermediates, and design decisions
Writing Composite Operations — step-by-step guide to building your own composites
CompositeDefinition Reference — API signatures and field tables
Error Handling in Practice — runtime failures, failure logs, and FailurePolicy
Advanced Patterns — customizing execution per step