Zenml Pipeline Authoring

Author ZenML Pipelines

This skill guides pipeline authoring: steps, artifacts, configuration, Docker settings, materializers, metadata, secrets, and visualizations.

Start Here: Interview the User

Do not rush to code. Before writing a single line, thoroughly understand what the user wants to build. The interview is the most important step — a well-scoped pipeline that does 3 things well beats a sprawling one that does 10 things poorly.

For complex or multi-pipeline projects: If the user describes something ambitious (e.g., "build me an end-to-end ML platform with data ingestion, feature engineering, training, evaluation, deployment, monitoring, and retraining"), or if they mention multiple pipelines, invoke the zenml-scoping skill first. It runs a deeper architectural interview that decomposes the system into pipeline units, identifies what doesn't belong in a pipeline at all, and produces a pipeline_architecture.md spec. Once that's done, come back here to build each pipeline one at a time.

For single, focused pipelines: If the user's request is clearly one pipeline (e.g., "build a training pipeline for my CSV data"), proceed with the questions below. If the answers are obvious from context, infer them and proceed. Only ask when genuinely ambiguous.

Q1: Static or dynamic pipeline? Most pipelines are static (fixed DAG). Use dynamic (@pipeline(dynamic=True)) only when the number of steps or their wiring depends on runtime values (e.g., "process N documents where N comes from a query"). See [Dynamic Pipelines](#dynamic-pipelines) and [references/dynamic-pipelines.md](references/dynamic-pipelines.md).

Q2: Local or remote orchestrator? If remote (Kubernetes, Vertex AI, SageMaker, AzureML), the [Artifact Golden Rule](#the-artifact-golden-rule) is critical, and you will need [Docker Settings](#docker-settings). If local-only for now, you can defer those concerns. Ask whether the user already has a stack set up — if not, point them to the ZenML docs for stack setup (this skill does not cover stack creation).

Q3: Any custom Python types? If steps produce or consume types beyond builtins, pandas, numpy, or Pydantic models, you likely need a [custom materializer](#custom-types-and-materializers). Note: Pydantic BaseModel subclasses have a built-in materializer — often the simplest alternative to writing a custom materializer.

Q4: Where should the project live? Ask the user where to create the project — a new subfolder, or the current directory. If the current directory is not empty, suggest a new subfolder.

Q5: What are the data sources? Understand where data comes from: local CSV/Parquet files, a database (Snowflake, PostgreSQL), an API, cloud storage? This determines the first step's implementation and whether secrets are needed. If credentials are involved, always use [ZenML Secrets](#secrets-management) — never pass passwords as CLI arguments or in config files.

Q6: Does the user want a small-data development mode? Many users want to iterate quickly with a fraction of the dataset. Plan for a --sample-size or --small CLI flag in run.py.

---

Core Anatomy

Defining steps

A step is a Python function decorated with @step. Type hints on inputs and outputs are required — they control serialization, caching, and dashboard display.

from zenml import step

@step
def train_model(X_train: pd.DataFrame, lr: float = 0.01) -> sklearn.base.BaseEstimator:
    """lr is a parameter (literal value); X_train is an artifact (from upstream step)."""
    model = LogisticRegression(C=1/lr).fit(X_train.drop("target", axis=1), X_train["target"])
    return model

Parameters vs artifacts: If a step input comes from another step's output, it is an artifact. If it is a literal value passed directly (JSON-serializable), it is a parameter. ZenML handles them differently.

Named and multi-output steps

Use Annotated to give outputs stable names. Use Tuple for multiple outputs:

from typing import Annotated, Tuple
from zenml import step
import pandas as pd

@step
def split_data(df: pd.DataFrame, ratio: float = 0.8) -> Tuple[
    Annotated[pd.DataFrame, "train"],
    Annotated[pd.DataFrame, "test"],
]:
    idx = int(len(df) * ratio)
    return df.iloc[:idx], df.iloc[idx:]

Wiring a pipeline

from zenml import pipeline

@pipeline
def training_pipeline(dataset_path: str = "data.csv", lr: float = 0.01) -> None:
    df = load_data(path=dataset_path)
    train, test = split_data(df=df)
    model = train_model(X_train=train, lr=lr)
    evaluate(model=model, X_test=test)

if __name__ == "__main__":
    training_pipeline()

Pipeline parameters (like dataset_path) can be overridden at runtime or via YAML config.

Step invocation IDs

When you call a step multiple times in one pipeline, ZenML auto-suffixes the name (scale, scale_2). Override with my_step(id="custom_id").

Project structure

Every pipeline project MUST follow this layout. This is non-negotiable — it produces clean, maintainable projects:

my_pipeline_project/
├── steps/                    # One file per step
│   ├── load_data.py
│   ├── preprocess.py
│   ├── train_model.py
│   └── evaluate.py
├── pipelines/
│   └── training.py           # Pipeline definition(s)
├── materializers/            # Custom materializers (if any)
│   └── my_data_materializer.py
├── visualizations/           # HTML/CSS templates for dashboard visualizations
│   └── metrics_report.html
├── configs/                  # One YAML config per environment
│   ├── dev.yaml
│   ├── staging.yaml
│   └── prod.yaml
├── run.py                    # CLI entry point (argparse, not click)
├── README.md                 # How to run, what stacks to use, etc.
└── pyproject.toml            # Dependencies — always pyproject.toml, not requirements.txt

Key rules:

• One step per file in a steps/ directory — not all steps in one steps.py.
• Separate pipeline definition from execution — pipeline in pipelines/, execution in run.py.
• Always create a README.md (not summary.md) explaining how to run the pipeline, what stacks it supports, and any setup needed. Link to the relevant ZenML docs pages (e.g., dynamic pipelines docs) rather than embedding lengthy explanations. Do NOT include stack registration or setup instructions — just say "assumes you have a ZenML stack configured" and link to https://docs.zenml.io for stack setup.
• Always use pyproject.toml for dependency declarations. Do NOT create requirements.txt alongside it — use one or the other, and pyproject.toml is the right choice.
• run.py uses argparse (not click) — click can conflict with ZenML's own click dependency.
• Run zenml init at the project root to set the source root explicitly — this prevents import failures when code runs inside containers.
• For uv/pyproject.toml package discovery, package data, and remote wheel inclusion pitfalls, see [references/runtime-portability-and-approvals.md](references/runtime-portability-and-approvals.md#uv-pyproject-and-package-discovery).

pyproject.toml template

Always use this as the starting point for pyproject.toml:

[project]
name = "my-pipeline-project"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "zenml>=0.93",
    "pandas>=2.0",
    # Add pipeline-specific dependencies here
]

[project.optional-dependencies]
dev = [
    "pytest",
    "ruff",
    "mypy",
]

Key version constraints:

• Python >= 3.12 — ZenML's modern features and type annotations benefit from 3.12+.
• ZenML >= 0.93 — this is the minimum for current features. For dynamic pipelines, require >= 0.91 at absolute minimum (but 0.93 is safer).
• Don't pin dev tool versions (pytest, ruff, mypy) — just list them without version constraints so users get the latest.
• Prefer uv in README instructions: uv pip install -e ".[dev]" for faster and more reliable resolution. If uv is unavailable in the user's environment, use pip install -e ".[dev]".

README template notes

The README should include:

• How to install: Prefer uv pip install -e ".[dev]" and zenml integration install --uv when supported. If uv is unavailable, use pip equivalents. Omit -y so users can review prompts.
• How to run: python run.py --config configs/dev.yaml
• What stacks it supports (just name them, don't explain how to register them)
• Link to specific orchestrator docs — not just generic https://docs.zenml.io. For example, if targeting Vertex AI, link to the Vertex AI orchestrator page and the GCP service connector page. Encourage users to use service connectors for authentication rather than manual credential management.
• A simple ASCII DAG visualization of the pipeline flow is a nice touch:

`` load_data --> preprocess --> train_model --> evaluate ``

run.py CLI template

Every run.py should offer these flags:

import argparse
from pipelines.training import training_pipeline

def main():
    parser = argparse.ArgumentParser(description="Run the training pipeline")
    parser.add_argument("--config", default="configs/dev.yaml", help="Path to YAML config")
    parser.add_argument("--no-cache", action="store_true", help="Disable caching")
    parser.add_argument("--sample-size", type=int, default=None,
                        help="Use only N rows (for quick local iteration)")
    args = parser.parse_args()

    pipeline_instance = training_pipeline.with_options(
        config_path=args.config,
        enable_cache=not args.no_cache,
    )
    pipeline_instance(sample_size=args.sample_size)

if __name__ == "__main__":
    main()

The sample_size parameter is passed as a pipeline parameter so the data-loading step can slice the dataset.

---

The Artifact Golden Rule

> Data must enter and move through the pipeline as artifacts, not as local file paths.

This is the single most important concept for cloud portability. When running on a remote orchestrator, each step runs in a separate container on a separate machine. There is no shared filesystem between steps.

What goes wrong

# ANTI-PATTERN: works locally, fails on cloud
@step
def preprocess(input_path: str) -> str:
    df = pd.read_csv(input_path)           # Reads from local disk
    output_path = "/tmp/processed.csv"
    df.to_csv(output_path)
    return output_path                      # Next step can't access /tmp on a different pod

@step
def train(data_path: str) -> None:
    df = pd.read_csv(data_path)             # FileNotFoundError on cloud!

The correct pattern

# CORRECT: data flows as artifacts
@step
def preprocess(input_path: str) -> pd.DataFrame:
    return pd.read_csv(input_path)          # ZenML serializes the DataFrame to the artifact store

@step
def train(data: pd.DataFrame) -> None:
    ...                                     # ZenML loads it from the artifact store — works everywhere

The first step in a pipeline is typically the one that bridges external data into the artifact world. All downstream steps receive artifacts, never file paths.

---

Dynamic Pipelines

Use dynamic pipelines when the DAG shape depends on runtime values. They are experimental and have restricted orchestrator support (Local, LocalDocker, Kubernetes, Vertex, SageMaker, AzureML). Always link to the dynamic pipelines documentation (https://docs.zenml.io/how-to/steps-pipelines/dynamic-pipelines) in the README since these APIs can be tricky to get right.

Minimal example

from zenml import pipeline, step

@step
def get_count() -> int:
    return 3

@step
def process(index: int) -> None:
    print(f"Processing {index}")

@pipeline(dynamic=True)
def my_dynamic_pipeline() -> None:
    count = get_count()
    count_data = count.load()       # .load() gets actual Python value
    for idx in range(count_data):
        process(index=idx)

The critical distinction: .load() vs .chunk()

| Method | Returns | Use for | |--------|---------|---------| | .load() | Actual Python data | Decisions, control flow, iteration | | .chunk(index=i) | A DAG edge reference | Wiring to downstream steps |

You typically need both: .load() to iterate/decide, .chunk() to wire the DAG:

items = produce_list()
for i, val in enumerate(items.load()):   # load to iterate
    if val > threshold:
        chunk = items.chunk(index=i)      # chunk to wire
        process(chunk)

Fan-out with .map() and parallel execution with .submit()

For map/reduce patterns, .map() fans out over a collection. For explicit parallelism, .submit() returns a future.

See [references/dynamic-pipelines.md](references/dynamic-pipelines.md) for the complete API: .map(), .product(), .submit(), unmapped(), .unpack(), child pipelines, .embed(), runtime modes, execution-mode caveats, orchestrator support table, and limitations.

---

Injecting External Data

When data originates outside the pipeline (a local file, a database, an API), you need to bridge it into the artifact system.

Pattern A: ExternalArtifact(value=...)

Upload data inline when defining the pipeline. Simple but disables caching for the consuming step:

from zenml import ExternalArtifact, pipeline, step
import pandas as pd

@step
def train(data: pd.DataFrame) -> None:
    ...

@pipeline
def my_pipeline() -> None:
    df = pd.read_csv("local_data.csv")
    train(data=ExternalArtifact(value=df))

Pattern B: Pre-upload + UUID reference (for remote orchestrators)

For dynamic pipelines on remote orchestrators, the pipeline function runs inside the orchestrator pod — it cannot read your local filesystem. Pre-upload the data, then reference it by UUID:

# run.py (client-side, runs on your machine)
from zenml.artifacts.utils import save_artifact
import pandas as pd

df = pd.read_csv("local_data.csv")
art = save_artifact(data=df, name="my_dataset")
print(art.id)  # Pass this UUID to the pipeline

# pipeline.py (runs inside the orchestrator pod)
from uuid import UUID
from zenml.client import Client

@pipeline
def my_pipeline(dataset_id: str) -> None:
    artifact = Client().get_artifact_version(UUID(dataset_id))
    train(data=artifact)

Important: do not construct ExternalArtifact(id=...) in user code. The public ExternalArtifact class is for value=... uploads; its internal config carries an ID only after upload. For existing artifacts, use Client().get_artifact_version(...). The stale patterns are ExternalArtifact(name=...), version=..., and model=....

See [references/external-data.md](references/external-data.md) for additional patterns including register_artifact().

---

YAML Configuration

Separate environment-specific settings from pipeline code using YAML config files.

Minimal example

# configs/dev.yaml
enable_cache: false
parameters:
  dataset_path: "data/small.csv"
  lr: 0.05
steps:
  train_model:
    settings:
      resources:
        cpu_count: 2

training_pipeline.with_options(config_path="configs/dev.yaml")()

Configuration precedence (highest to lowest): Runtime Python code > Step-level YAML > Pipeline-level YAML > Defaults.

Always use separate config files per environment (configs/dev.yaml, configs/staging.yaml, configs/prod.yaml) — never a single config.yaml. Generate a template with zenml pipeline build-configuration my_pipeline > config_template.yaml.

Prefer with_options() (returns a copy) over configure() (mutates in place).

See [references/yaml-config.md](references/yaml-config.md) for the complete YAML schema and multi-env pattern.

---

Docker Settings

When running on remote orchestrators, ZenML builds Docker images for each step. Use DockerSettings to control what goes into those images.

Common patterns

from zenml.config import DockerSettings

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [zenml-io](https://github.com/zenml-io)
- **Source:** [zenml-io/skills](https://github.com/zenml-io/skills)
- **License:** MIT
- **Homepage:** https://docs.zenml.io

Install and usage instructions live in the source repository linked above.