Branching and Merging

What you’ll learn¶

Split a data stream into parallel branches (fan-out)
Produce multiple output variants from a single input
Recombine branches with Merge (passthrough semantics)
Build a complete branch-and-reconverge pipeline

Prerequisites: Sources and Sequencing
Estimated time: 10 minutes

The previous tutorial covered linear pipelines. Real workflows often need to explore alternatives in parallel — apply two different transformations to the same data, then pick the best results. This tutorial covers the three patterns that make that possible.

Pattern	What it does
Branching	Route one step’s output to multiple downstream steps
Variants	One input produces multiple outputs within a single step
Merge	Combine artifact streams back into one

from __future__ import annotations

from artisan.operations.curator import Merge
from artisan.operations.examples import DataGenerator, DataTransformer
from artisan.orchestration import PipelineManager
from artisan.utils import tutorial_setup
from artisan.visualization import build_macro_graph, build_micro_graph

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

Branching¶

Branching is the most natural pattern: take one step’s output and feed it to two (or more) downstream steps. Each branch receives the same artifact IDs as input but processes them independently.

This is useful whenever you want to compare alternative processing strategies — for example, running two different transformation seeds to see which produces better results.

env_branch = tutorial_setup("branching")

pipeline = PipelineManager.create(
    name="branching",
    delta_root=env_branch.delta_root,
    staging_root=env_branch.staging_root,
    working_root=env_branch.working_root,
)
output = pipeline.output

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

# Both branches consume the same output — they just use different params
pipeline.run(
    operation=DataTransformer,
    inputs={"dataset": output("generate", "datasets")},
    params={"seed": 100},
    name="transform_a",
)
pipeline.run(
    operation=DataTransformer,
    inputs={"dataset": output("generate", "datasets")},
    params={"seed": 200},
    name="transform_b",
)

result = pipeline.finalize()

build_macro_graph(env_branch.delta_root)

The generate step feeds into both transform steps. Each branch operates on the same two input artifacts but produces independent outputs. There is no special “branch” API — you create a branch by passing the same OutputReference to multiple pipeline.run() calls.

build_micro_graph(env_branch.delta_root)

Variants (in-step fan-out)¶

Branching splits the pipeline graph. Variants split within a single step: one input artifact produces multiple output artifacts. The variants parameter on DataTransformer controls this — each variant gets a different random noise pattern.

Branching	Variants
Multiple steps, same inputs	One step, multiple outputs per input
Different operations or params per branch	Same operation, automatic variation
Separate output streams	Single output stream (all variants together)

env_variants = tutorial_setup("variants")

pipeline = PipelineManager.create(
    name="variants",
    delta_root=env_variants.delta_root,
    staging_root=env_variants.staging_root,
    working_root=env_variants.working_root,
)
output = pipeline.output

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

# Each of the 2 input datasets produces 3 variants = 6 total outputs
pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("generate", "datasets")},
    params={"variants": 3, "seed": 100},
)

result = pipeline.finalize()

build_macro_graph(env_variants.delta_root)

The transform step consumed 2 artifacts and produced 6 (3 variants each). All outputs live in the same step and the same output role. The provenance graph tracks which variant came from which input via lineage edges (orange arrows).

build_micro_graph(env_variants.delta_root)

Merge¶

Merge combines multiple artifact streams into one. It uses passthrough semantics — no new artifacts are created. The output is the union of all input artifact IDs, which means downstream steps see the original artifacts from every branch as if they came from a single source.

Two things to know about Merge:

All input streams must contain the same artifact type.
The merged output role is always named "merged".

env_merge = tutorial_setup("merge")

pipeline = PipelineManager.create(
    name="merge",
    delta_root=env_merge.delta_root,
    staging_root=env_merge.staging_root,
    working_root=env_merge.working_root,
)
output = pipeline.output

pipeline.run(operation=DataGenerator, name="gen_a", params={"count": 2, "seed": 42})
pipeline.run(operation=DataGenerator, name="gen_b", params={"count": 3, "seed": 99})

# Merge accepts arbitrary role names — they're just labels
pipeline.run(
    operation=Merge,
    name="merge",
    inputs={
        "a": output("gen_a", "datasets"),
        "b": output("gen_b", "datasets"),
    },
)

result = pipeline.finalize()

build_macro_graph(env_merge.delta_root)

The Merge step has no output artifacts of its own — it unions the 2 + 3 = 5 artifacts from its inputs. Downstream steps that consume output("merge", "merged") would receive all 5 artifacts.

build_micro_graph(env_merge.delta_root)

Merge scales to any number of input streams. The input role names ("a", "b", "c", ...) are arbitrary labels — pick whatever is meaningful for your pipeline.

Putting it together: branch and reconverge¶

The most common real-world pattern combines branching and merge: generate data, branch into parallel transformations, merge the results back together, then continue processing the combined stream.

                ┌─ transform_a ─┐
  generate ─────┤               ├─── merge ─── downstream
                └─ transform_b ─┘

env_roundtrip = tutorial_setup("roundtrip")

pipeline = PipelineManager.create(
    name="roundtrip",
    delta_root=env_roundtrip.delta_root,
    staging_root=env_roundtrip.staging_root,
    working_root=env_roundtrip.working_root,
)
output = pipeline.output

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

# Branch into two transformation strategies
pipeline.run(
    operation=DataTransformer,
    name="transform_a",
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 0.5, "seed": 100},
)
pipeline.run(
    operation=DataTransformer,
    name="transform_b",
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 2.0, "seed": 200},
)

# Merge the branches back together
pipeline.run(
    operation=Merge,
    name="merge",
    inputs={
        "small": output("transform_a", "dataset"),
        "large": output("transform_b", "dataset"),
    },
)

# Process the combined stream
pipeline.run(
    operation=DataTransformer,
    name="final_transform",
    inputs={"dataset": output("merge", "merged")},
    params={"noise_amplitude": 0.01, "seed": 300},
)

result = pipeline.finalize()

build_macro_graph(env_roundtrip.delta_root)

The graph shows the diamond pattern: generate → branch → merge → downstream. The final step receives all 4 artifacts (2 from each branch) as a single input stream. Provenance tracks which original artifacts each final output descends from, across the branch boundaries.

build_micro_graph(env_roundtrip.delta_root)

Summary¶

This tutorial covered three patterns for non-linear pipelines:

Branching — pass the same OutputReference to multiple steps. No special API, the graph topology emerges from how you wire inputs.
Variants — use the variants parameter to produce multiple outputs per input within a single step.
Merge — combine streams with passthrough semantics. No new artifacts are created; downstream steps see the union of all inputs.

Next steps¶

Metrics and Filtering — score artifacts with metrics, then filter by threshold
Operations Model — how creators and curators differ under the hood
Execution Flow — what happens when pipeline.run() executes a step