Skip to main content

​
Overview

The Stack class represents a complete ZenML stack - a collection of stack components that together define the infrastructure and tooling needed to run ML pipelines. Each stack consists of required components (orchestrator, artifact store) and optional components (container registry, experiment tracker, etc.).

​
Class Definition

from zenml.stack import Stack

​
Stack Components

A stack is composed of the following components:

​
Required Components

  • Orchestrator: Coordinates pipeline execution
  • Artifact Store: Stores pipeline artifacts and outputs

​
Optional Components

  • Container Registry: Stores Docker images for containerized execution
  • Step Operator: Executes individual steps on specialized infrastructure
  • Experiment Tracker: Tracks experiments and metrics
  • Model Deployer: Deploys trained models
  • Feature Store: Manages and serves features
  • Alerter: Sends alerts and notifications
  • Annotator: Manages data annotation
  • Data Validator: Validates data quality
  • Image Builder: Builds Docker images
  • Model Registry: Manages model versions
  • Deployer: Deploys pipelines to production
  • Log Store: Stores and manages logs

​
Initialization

def __init__(
    self,
    id: UUID,
    name: str,
    *,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    orchestrator: BaseOrchestrator,
    artifact_store: BaseArtifactStore,
    container_registry: Optional[BaseContainerRegistry] = None,
    # ... other optional components
):
​
id
UUID
required
Unique ID of the stack.
​
name
str
required
Name of the stack.
​
environment
Dict[str, str]
Environment variables to set when running on this stack.
​
secrets
List[UUID]
Secrets to make available as environment variables.
​
orchestrator
BaseOrchestrator
required
The orchestrator component.
​
artifact_store
BaseArtifactStore
required
The artifact store component.
​
container_registry
BaseContainerRegistry
Optional container registry component.
​
step_operator
BaseStepOperator
Optional step operator component.
​
experiment_tracker
BaseExperimentTracker
Optional experiment tracker component.
​
model_deployer
BaseModelDeployer
Optional model deployer component.

​
Class Methods

​
from_model

@classmethod
def from_model(cls, stack_model: StackResponse) -> Stack:
    """Creates a Stack instance from a StackResponse."""
​
stack_model
StackResponse
required
The stack model to create the Stack from.
​
return
Stack
The created Stack instance.
This method includes caching to avoid redundant database calls for the same stack.

​
from_components

@classmethod
def from_components(
    cls,
    id: UUID,
    name: str,
    components: Dict[StackComponentType, StackComponent],
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
) -> Stack:
    """Creates a stack from a dictionary of components."""
​
id
UUID
required
Unique ID of the stack.
​
name
str
required
Name of the stack.
​
components
Dict[StackComponentType, StackComponent]
required
Dictionary mapping component types to component instances.
​
return
Stack
A stack instance consisting of the given components.

​
Properties

​
id

@property
def id(self) -> UUID:
    """The ID of the stack."""

​
name

@property
def name(self) -> str:
    """The name of the stack."""

​
components

@property
def components(self) -> Dict[StackComponentType, StackComponent]:
    """All components of the stack."""
​
return
Dict[StackComponentType, StackComponent]
Dictionary of all stack components, keyed by component type.

​
orchestrator

@property
def orchestrator(self) -> BaseOrchestrator:
    """The orchestrator of the stack."""

​
artifact_store

@property
def artifact_store(self) -> BaseArtifactStore:
    """The artifact store of the stack."""

​
container_registry

@property
def container_registry(self) -> Optional[BaseContainerRegistry]:
    """The container registry of the stack."""

​
requires_remote_server

@property
def requires_remote_server(self) -> bool:
    """Whether the stack requires a remote ZenML server."""
​
return
bool
True if any code executes remotely (remote orchestrator or step operator).

​
environment

@property
def environment(self) -> Dict[str, str]:
    """Environment variables to set when running on this stack."""

​
secrets

@property
def secrets(self) -> List[UUID]:
    """Secrets to set as environment variables."""

​
Core Methods

​
validate

def validate(self, fail_if_secrets_missing: bool = False) -> None:
    """Validates the stack configuration."""
​
fail_if_secrets_missing
bool
default:"False"
If True, raises an error if secrets are missing. Otherwise logs a warning.
Validation checks:
  • Stack has required image builder if components need it
  • All component validators pass
  • Required secrets exist

​
submit_pipeline

def submit_pipeline(
    self,
    snapshot: PipelineSnapshotResponse,
    placeholder_run: Optional[PipelineRunResponse] = None,
) -> None:
    """Submits a pipeline for execution on this stack."""
​
snapshot
PipelineSnapshotResponse
required
The pipeline snapshot to submit.
​
placeholder_run
PipelineRunResponse
Optional placeholder run for the pipeline.

​
deploy_pipeline

def deploy_pipeline(
    self,
    snapshot: PipelineSnapshotResponse,
    deployment_name: str,
    timeout: Optional[int] = None,
) -> DeploymentResponse:
    """Deploys a pipeline on this stack."""
​
snapshot
PipelineSnapshotResponse
required
The pipeline snapshot to deploy.
​
deployment_name
str
required
Name for the deployment.
​
timeout
int
Maximum time in seconds to wait for deployment.
​
return
DeploymentResponse
The deployment response.

​
prepare_step_run

def prepare_step_run(self, info: StepRunInfo) -> None:
    """Prepares running a step."""
​
info
StepRunInfo
required
Information about the step that will be executed.

​
cleanup_step_run

def cleanup_step_run(self, info: StepRunInfo, step_failed: bool) -> None:
    """Cleans up resources after step execution."""
​
info
StepRunInfo
required
Information about the step that was executed.
​
step_failed
bool
required
Whether the step failed.

​
Utility Methods

​
requirements

def requirements(
    self,
    exclude_components: Optional[AbstractSet[StackComponentType]] = None,
) -> Set[str]:
    """Get PyPI requirements for the stack."""
​
exclude_components
AbstractSet[StackComponentType]
Component types to exclude from requirements.
​
return
Set[str]
Set of PyPI package requirements.

​
apt_packages

@property
def apt_packages(self) -> List[str]:
    """List of APT package requirements for the stack."""

​
get_docker_builds

def get_docker_builds(
    self,
    snapshot: PipelineSnapshotBase
) -> List[BuildConfiguration]:
    """Gets Docker builds required for the stack."""
​
snapshot
PipelineSnapshotBase
required
The pipeline snapshot for which to get builds.
​
return
List[BuildConfiguration]
List of required Docker build configurations.

​
get_step_run_metadata

def get_step_run_metadata(
    self,
    info: StepRunInfo
) -> Dict[UUID, Dict[str, MetadataType]]:
    """Get component-specific metadata for a step run."""
​
info
StepRunInfo
required
Information about the step that was executed.
​
return
Dict[UUID, Dict[str, MetadataType]]
Dictionary mapping component IDs to their metadata.

​
Usage Example

from zenml.client import Client
from zenml.stack import Stack

# Get the active stack
client = Client()
active_stack = client.active_stack

print(f"Stack name: {active_stack.name}")
print(f"Orchestrator: {active_stack.orchestrator.name}")
print(f"Artifact store: {active_stack.artifact_store.name}")

# Check if stack has certain components
if active_stack.experiment_tracker:
    print(f"Experiment tracker: {active_stack.experiment_tracker.name}")

if active_stack.container_registry:
    print(f"Container registry: {active_stack.container_registry.name}")

# Validate stack
active_stack.validate(fail_if_secrets_missing=True)

# Get all components
for component_type, component in active_stack.components.items():
    print(f"{component_type.value}: {component.name}")

# Check stack requirements
print(f"Requires remote server: {active_stack.requires_remote_server}")
print(f"PyPI requirements: {active_stack.requirements()}")
print(f"APT packages: {active_stack.apt_packages}")

​
Creating a Custom Stack

from uuid import uuid4
from zenml.client import Client
from zenml.enums import StackComponentType

client = Client()

# Create a new stack from existing components
stack = client.create_stack(
    name="my_custom_stack",
    components={
        StackComponentType.ORCHESTRATOR: "local_orchestrator",
        StackComponentType.ARTIFACT_STORE: "local_artifact_store",
        StackComponentType.CONTAINER_REGISTRY: "my_registry",
        StackComponentType.EXPERIMENT_TRACKER: "mlflow_tracker",
    }
)

# Set as active stack
client.activate_stack(stack.id)

# Verify
print(f"Active stack: {client.active_stack_model.name}")

​
Validation Example

from zenml.stack import Stack
from zenml.exceptions import StackValidationError

try:
    stack.validate(fail_if_secrets_missing=True)
    print("Stack validation passed!")
except StackValidationError as e:
    print(f"Stack validation failed: {e}")
    # Handle validation error

​
See Also

StackComponent

Learn about individual stack components

BaseOrchestrator

Understand the orchestrator component

Stacks Guide

Complete guide to ZenML stacks

Stack Switching

Learn how to switch between stacks

Build docs developers (and LLMs) love

Get started for freeTalk to us