Your First Pipeline

What you’ll learn¶

Create a pipeline and run operations step by step
Wire step outputs to step inputs using output references
Merge two data streams and transform the result
Compute metrics and filter artifacts by score
Inspect the pipeline summary and provenance graph

Prerequisites: Pixi installed, Prefect server running (see below). Estimated time: 15 minutes GPU required: No.

An Artisan pipeline is a sequence of steps. Each step runs an operation that consumes input artifacts and produces output artifacts. You build the pipeline by adding steps one at a time, wiring each step’s inputs to a previous step’s outputs.

By the end of this tutorial you’ll have built and run this pipeline:

IngestData   ──┐
                ├── Merge ── DataTransformer ── MetricCalculator
DataGenerator ──┘                                     │
                                                Filter (median > 0.15)
                                                      │
                                              DataTransformer

Every artifact the pipeline produces is tracked with full provenance — you can always trace any output back to the inputs and parameters that created it.

Setup¶

Artisan uses a Prefect server to dispatch work — even for local execution. Start one before running this notebook:

pixi run prefect-start

This launches the server in the background and writes a discovery file so Artisan can find it automatically. No environment variables or extra configuration needed. To stop it later:

pixi run prefect-stop

See the Installation guide for details, including how to connect to an existing server.

from __future__ import annotations

from artisan.operations.curator import Filter, IngestData, Merge
from artisan.operations.examples import (
    DataGenerator,
    DataTransformer,
    MetricCalculator,
)
from artisan.orchestration import PipelineManager
from artisan.utils import find_project_root, tutorial_setup
from artisan.visualization import inspect_pipeline

tutorial_setup creates a clean directory tree for this tutorial’s data (Delta Lake tables, staging files, and working files). find_project_root locates the repository root so we can reference test fixtures.

PROJECT_ROOT = find_project_root()
SOURCE_FILES = sorted((PROJECT_ROOT / "tests" / "fixtures" / "csv").glob("*.csv"))[:2]

env = tutorial_setup("first_pipeline")
print(f"Delta root:   {env.delta_root}")
print(f"Source files:  {[f.name for f in SOURCE_FILES]}")

Delta root:   /Users/andrewhunt/git/artisan-dev/docs/tutorials/getting-started/runs/first_pipeline/delta
Source files:  ['d0.csv', 'd1.csv']

Create a pipeline¶

A PipelineManager coordinates step execution and stores all artifacts in Delta Lake tables. You create one by giving it a name and the directory paths from the tutorial environment.

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

PipelineManager(name='first_pipeline', steps=0, delta_root=PosixPath('/Users/andrewhunt/git/artisan-dev/docs/tutorials/getting-started/runs/first_pipeline/delta'))

Step 0 — Ingest CSV files¶

Every pipeline starts with data. IngestData brings external files into the pipeline’s tracking system. Pass a list of file paths as inputs.

pipeline.run(
    operation=IngestData,
    name="ingest",
    inputs=[str(f) for f in SOURCE_FILES],
)

StepResult(step_name='ingest', step_number=0, success=True, total_count=2, succeeded_count=2, failed_count=0, output_roles=frozenset({'data'}), output_types={'data': 'data'}, duration_seconds=0.2597384590044385, metadata={'timings': {'resolve_inputs': 0.0, 'batch_and_cache': 0.0, 'execute': 0.2314, 'verify_staging': 0.0, 'commit': 0.0199, 'compact': 0.0082, 'total': 0.2596}})

pipeline.run() executes the operation and returns a StepResult. The name parameter gives this step a human-readable label we’ll use later to reference its outputs.

The result shows the step succeeded and produced 2 data artifacts (one per CSV file) plus 2 file references.

Step 1 — Generate synthetic data¶

DataGenerator creates CSV datasets from scratch — no inputs needed, only parameters. This is useful for testing and prototyping.

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

07:02:07.802 | WARNING | EventsWorker - Still processing items: 3 items remaining...

We now have two independent data streams: 2 artifacts from ingestion and 2 from generation. Next, we’ll combine them.

Step 2 — Merge streams¶

Merge unions multiple data streams into one. It uses passthrough semantics — no new artifacts are created, the existing ones are routed into a single output stream.

This is where output references come in. To tell Merge where its inputs come from, we use pipeline.output(step_name, role) to reference a previous step’s output by name and role.

pipeline.run(
    operation=Merge,
    name="merge",
    inputs={
        "branch_a": output("ingest", "data"),
        "branch_b": output("generate", "datasets"),
    },
)

The inputs dictionary maps input role names (which you choose) to output references from earlier steps. Each output reference says: “give me the artifacts from step X with role Y.”

output("ingest", "data") → the 2 data artifacts from step 0
output("generate", "datasets") → the 2 datasets from step 1

After merging, the "merged" output role contains all 4 artifacts.

Step 3 — Transform¶

DataTransformer reads each input CSV, scales the numeric columns by scale_factor, and writes the result. It processes each artifact independently.

pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("merge", "merged")},
    params={"scale_factor": 0.5, "variants": 1, "seed": 100},
)

Step 4 — Compute metrics¶

MetricCalculator computes distribution statistics (min, max, median, range) from the score column of each dataset. It produces one metric artifact per input.

pipeline.run(
    operation=MetricCalculator,
    name="metrics",
    inputs={"dataset": output("transform", "dataset")},
)

Step 5 — Filter by score¶

Filter evaluates criteria against each artifact’s metrics and passes through only those that meet the threshold. Like Merge, Filter uses passthrough semantics — it doesn’t create new artifacts, it selects from existing ones.

Filter auto-discovers metrics via forward provenance walk from the passthrough artifacts. Criteria use bare field names — "distribution.median" means the distribution.median key in whatever metric artifacts are descendants of the passthrough artifacts.

Here we keep only datasets where the median score exceeds 0.15.

pipeline.run(
    operation=Filter,
    name="filter",
    inputs={"passthrough": output("transform", "dataset")},
    params={
        "criteria": [
            {"metric": "distribution.median", "operator": "gt", "value": 0.15},
        ]
    },
)

The step result shows how many artifacts passed and how many were rejected. Notice that Filter doesn’t need an explicit reference to the metrics step — it finds the relevant metric artifacts automatically by walking the provenance graph forward from each passthrough artifact.

Step 6 — Final transformation¶

Run one more transformation on the filtered artifacts. This step only processes artifacts that passed the filter.

pipeline.run(
    operation=DataTransformer,
    name="refine",
    inputs={"dataset": output("filter", "passthrough")},
    params={"scale_factor": 0.1, "variants": 1, "seed": 101},
)

Finalize¶

Call finalize() to close the pipeline. This waits for any pending work and returns a summary.

result = pipeline.finalize()
print(
    f"Pipeline complete: {result['total_steps']} steps, success={result['overall_success']}"
)

Inspect results¶

Now that the pipeline has run, let’s see what it produced.

Pipeline summary¶

inspect_pipeline shows one row per step: the operation, status, what it produced, and how long it took.

inspect_pipeline(env.delta_root)

A few things to notice:

Merge produced no new artifacts (the produced column shows -). It routed existing artifacts without copying them.
Filter shows how many artifacts “passed” rather than artifact counts. Like Merge, it works by selecting, not creating.
Refine only processes the artifacts that survived the filter.

Summary¶

Here’s what we covered:

Concept	What it does
`PipelineManager.create()`	Creates a new pipeline with storage paths
`pipeline.run(operation=Op, ...)`	Runs an operation as the next step
`output = pipeline.output`	Binds the output reference helper for concise wiring
`output(name, role)`	References a previous step’s output by name
`pipeline.finalize()`	Closes the pipeline and returns a summary
`name="..."` on `run()`	Gives a step a name for later reference
`params={...}` on `run()`	Passes parameters to the operation

The two categories of operations we used:

Creator operations (IngestData, DataGenerator, DataTransformer, MetricCalculator) produce new artifacts
Passthrough operations (Merge, Filter) route or select existing artifacts without creating new ones

Every artifact is tracked in Delta Lake tables with full provenance. The same inputs and parameters always produce the same artifact IDs (content addressing), which means re-running a pipeline can skip steps whose inputs haven’t changed.

Next steps¶

Exploring Results — Dig into Delta tables, query lineage, and inspect individual artifacts
Run vs Submit — Blocking vs async execution modes
Pipeline Patterns — Reusable pipeline topologies (branching, fan-out, cycles)
Provenance System — How artifact tracking and content addressing work under the hood
Storage and Delta Lake — Where artifacts are stored and how to query them