Sources and Sequencing

What you’ll learn¶

Start a pipeline by generating data from scratch (generative source)
Start a pipeline by ingesting existing files (external ingest)
Chain steps together so each consumes the previous step’s output
Read provenance graphs to trace data flow

Prerequisites: Your First Pipeline
Estimated time: 10 minutes

Every pipeline starts somewhere. This tutorial covers the two ways to get data into a pipeline and the most common way to move it forward: a linear chain.

from __future__ import annotations

from artisan.operations.curator import IngestData
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 (
    build_macro_graph,
    build_micro_graph,
    inspect_metrics,
    inspect_pipeline,
)

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

Reading the provenance graphs¶

Each example ends with a provenance graph. Here’s how to read them:

Grey boxes are execution steps (one per pipeline.run() call)
Blue boxes are artifacts produced by each step
Arrows show data flow: which step consumed which artifacts, and which step produced them

Pattern 1: Generative source¶

A generative source creates artifacts from scratch — no inputs required. This is the smallest possible pipeline: one step, zero inputs.

env = tutorial_setup("sources_generative")

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

pipeline.run(operation=DataGenerator, name="generate", params={"count": 3, "seed": 42})
pipeline.finalize()  # seal the run and write final provenance metadata

inspect_pipeline(env.delta_root)

DataGenerator produced 3 data artifacts from nothing. The count parameter controls how many datasets are created.

build_macro_graph(env.delta_root)

build_micro_graph(env.delta_root)

One step, three outputs, no input arrows. This is the generative source pattern — data flows only outward from the step.

Pattern 2: External file ingest¶

When your data already exists on disk, IngestData brings it into the provenance graph. Pass file paths as inputs — the framework promotes them to FileRefArtifacts, then reads their content into DataArtifacts.

env = tutorial_setup("sources_ingest")

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

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

inspect_pipeline(env.delta_root)

Two input files produced two pairs of artifacts: a data artifact (loaded CSV content) and a file_ref artifact (metadata reference to the original path) for each file.

build_macro_graph(env.delta_root)

build_micro_graph(env.delta_root)

The graph shows ingest with input arrows (the file references it consumed) and output arrows (the data artifacts it produced). Both artifact types are tracked with full provenance, so you can always trace a dataset back to the file it came from.

Pattern 3: Linear chain¶

A chain connects steps sequentially: each step consumes the previous step’s output. You wire them together with pipeline.output("step_name", "role"), which creates an OutputReference — a lazy pointer that the framework resolves at execution time.

For readability, the examples below assign output = pipeline.output so wiring calls read as output("step_name", "role").

This example builds a three-step chain:

DataGenerator  →  DataTransformer  →  MetricCalculator
 (generate)        (transform)          (metrics)

env = tutorial_setup("sources_chain")

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

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

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

# Step 2: compute distribution metrics
pipeline.run(
    operation=MetricCalculator,
    name="metrics",
    inputs={"dataset": output("transform", "dataset")},
)

pipeline.finalize()

The key line is inputs={"dataset": output("generate", "datasets")}. This tells the framework: “for this step’s dataset input role, use whatever the generate step produced under the datasets output role.” The framework resolves this at dispatch time by querying the Delta Lake tables.

inspect_pipeline(env.delta_root)

Two datasets flowed through the entire chain. Each step consumed the previous step’s output and produced its own artifacts.

inspect_metrics(env.delta_root, step_number=2)

MetricCalculator computed distribution statistics (min, max, median, range) and summary statistics (coefficient of variation, row count) for each transformed dataset.

build_macro_graph(env.delta_root)

build_micro_graph(env.delta_root)

Three steps connected left to right. Arrows between steps show data flow: step 0 produced datasets, step 1 consumed and transformed them, step 2 computed metrics from the transformed outputs.

Combining patterns: ingest into a chain¶

Sources and chains compose naturally. Here, IngestData feeds external files into a transform-then-score chain — the same wiring pattern as before, with a different starting point.

env = tutorial_setup("sources_ingest_chain")

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

# Step 0: ingest CSV files from disk
pipeline.run(operation=IngestData, name="ingest", inputs=[str(f) for f in SOURCE_FILES])

# Step 1: transform the ingested data
pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("ingest", "data")},
    params={"scale_factor": 2.0, "seed": 42},
)

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

pipeline.finalize()

inspect_pipeline(env.delta_root)

build_macro_graph(env.delta_root)

build_micro_graph(env.delta_root)

The graph now starts with an ingest step instead of a generator, but the downstream chain is identical. The only difference in code is the first pipeline.run() call and the output role name ("data" for IngestData vs. "datasets" for DataGenerator).

Summary¶

You learned three patterns for getting data into and through a pipeline:

Pattern	When to use	Key idea
Generative source	Data is created from scratch	No inputs, operation produces artifacts directly
External ingest	Data already exists on disk	File paths are promoted to tracked artifacts
Linear chain	Sequential processing	`output("step_name", "role")` wires steps together lazily

Next steps¶

Branching and Merging — fan-out to multiple branches, then recombine with Merge
Execution Flow — how the framework resolves OutputReferences and dispatches work
Building a Pipeline — task-oriented guide for common pipeline construction patterns