Composites and Composition

Operations are the unit of computation. Composites are the unit of reuse. When operations are tightly coupled — when you always run transform-then-score, or preprocess-then-analyze — extracting each pair as a separate pipeline step wastes I/O on intermediate artifacts that are immediately consumed. A composite solves this by grouping operations into a reusable unit with declared inputs, outputs, and internal wiring.

This page explains what composites are, why they exist, how they execute, and when to use them.

The problem composites solve¶

Consider three approaches to running two tightly coupled operations:

A composite encapsulates the wiring once. Every pipeline that uses it gets the same internal structure, the same parameter forwarding, and the same output contract — without duplicating code.

Anatomy of a CompositeDefinition¶

A composite is a subclass of CompositeDefinition. It looks similar to an OperationDefinition in structure — name, InputRole, OutputRole, inputs, outputs — but instead of implementing a computation lifecycle, it implements compose().

class TransformAndScore(CompositeDefinition):
    name = "transform_and_score"

    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"),
    }

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

The key difference from an operation: compose does not compute. It wires. Each ctx.run() delegates to a real operation. The composite itself produces no artifacts — it orchestrates the operations that do.

compose() vs the operation lifecycle¶

How compose() wires operations¶

compose() receives a CompositeContext with three methods:

ctx.input(role) → CompositeRef¶

Reference a declared input of the composite. The returned CompositeRef is passed as an input to ctx.run():

dataset_ref = ctx.input("dataset")

ctx.run(operation, ...) → CompositeStepHandle¶

Execute an operation (or nested composite). Returns a handle whose .output(role) method produces a CompositeRef for wiring to the next operation:

handle = ctx.run(DataTransformer, inputs={"dataset": dataset_ref})
transformed_ref = handle.output("dataset")

ctx.output(role, ref)¶

Map an internal result to a declared output of the composite:

ctx.output("metrics", scored.output("metrics"))

Only refs that are mapped via ctx.output() are visible outside the composite. Everything else is an intermediate.

Collapsed vs expanded execution¶

The same composite definition supports two execution modes. The pipeline caller chooses.

When to use each¶

Collapsed when:

• Internal operations are always run together

• Intermediate artifacts are not needed after the composite completes

• You want to minimize Delta Lake I/O

• The combined computation fits on a single worker

Expanded when:

• Internal operations have different resource requirements (e.g., GPU vs CPU)

• You want independent caching per operation

• You need to inspect intermediate results as first-class pipeline steps

• Operations can benefit from independent parallelism

How it works¶

Collapsed:
  Worker
  ┌────────────────────────────────────────┐
  │  ctx.run(OpA) ──in-memory──▶ ctx.run(OpB)  │
  │         │                         │         │
  │         ▼                         ▼         │
  │    intermediates            declared outputs │
  │    (configurable)           (committed)      │
  └────────────────────────────────────────┘

Expanded:
  Step N                    Step N+1
  ┌──────────────┐         ┌──────────────┐
  │  OpA         │──Delta──│  OpB         │
  │  (dispatch,  │  Lake   │  (dispatch,  │
  │   execute,   │         │   execute,   │
  │   commit)    │         │   commit)    │
  └──────────────┘         └──────────────┘

Intermediate artifact handling¶

In collapsed mode, the intermediates parameter controls what happens to artifacts produced by non-final internal operations:

pipeline.run(
    operation=TransformAndScore,
    inputs={"dataset": output("gen", "datasets")},
    intermediates="persist",  # keep intermediates for debugging
)

In expanded mode, intermediates are always full pipeline steps — each gets its own Delta Lake commit, provenance edges, and cache entry.

Nesting composites¶

A composite can contain other composites. ctx.run() accepts both OperationDefinition and CompositeDefinition subclasses:

class GenerateAndScore(CompositeDefinition):
    name = "generate_and_score"
    # ...

    def compose(self, ctx: CompositeContext) -> None:
        generated = ctx.run(DataGenerator, params={"count": 3})
        scored = ctx.run(
            TransformAndScore,  # nested composite
            inputs={"dataset": generated.output("datasets")},
        )
        ctx.output("metrics", scored.output("metrics"))

In collapsed mode, nested composites execute recursively on the same worker. In expanded mode, each nested composite’s internal operations become their own pipeline steps (step names are dot-separated: outer.inner.operation).

Relationship to operations¶

A composite is not a subclass of OperationDefinition. It does not compute — it composes. The two share structural similarities (InputRole, OutputRole, inputs, outputs) because both need to declare their data contract, but they are distinct abstractions:

Operations are leaves. Composites are branches. Both are nodes in the pipeline DAG.

Key design decisions¶

Cross-references¶

• CompositeDefinition Reference — API signatures and field tables

• Writing Composite Operations — step-by-step guide

• Composable Operations Tutorial — interactive examples

• Operations Model — the operation abstractions that composites compose

• Execution Flow — how collapsed and expanded composites fit into the dispatch-execute-commit lifecycle

• Architecture Overview — where composites sit in the five-layer architecture