StackComponent

Overview

The StackComponent class is the abstract base class for all stack components in ZenML. Stack components are modular building blocks that provide specific functionality within a ZenML stack, such as orchestration, artifact storage, experiment tracking, and more.

Class Definition

from zenml.stack import StackComponent

Component Types

ZenML supports the following stack component types:

ORCHESTRATOR - Pipeline execution coordination
ARTIFACT_STORE - Artifact and output storage
CONTAINER_REGISTRY - Docker image storage
STEP_OPERATOR - Specialized step execution
EXPERIMENT_TRACKER - Experiment tracking and logging
MODEL_DEPLOYER - Model deployment
FEATURE_STORE - Feature management
ALERTER - Notifications and alerts
ANNOTATOR - Data annotation
DATA_VALIDATOR - Data quality validation
IMAGE_BUILDER - Docker image building
MODEL_REGISTRY - Model version management
DEPLOYER - Pipeline deployment
LOG_STORE - Log storage and management

Initialization

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
):

name

str

required

The name of the component.

UUID

required

Unique ID of the component.

config

StackComponentConfig

required

Configuration for the component.

flavor

str

required

The flavor of the component (e.g., “local”, “kubernetes”, “aws”).

type

StackComponentType

required

The type of the component.

user

UUID

ID of the user who created the component.

created

datetime

required

Creation timestamp.

updated

datetime

required

Last update timestamp.

environment

Dict[str, str]

Environment variables to set when using this component.

secrets

List[UUID]

Secrets to make available as environment variables.

labels

Dict[str, Any]

Labels for organizing components.

connector_requirements

ServiceConnectorRequirements

Requirements for service connector integration.

connector

UUID

ID of a linked service connector.

connector_resource_id

str

Custom resource ID to access through the connector.

Class Methods

from_model

@classmethod
def from_model(cls, component_model: ComponentResponse) -> StackComponent:
    """Creates a StackComponent from a ComponentResponse."""

component_model

ComponentResponse

required

The component model to create the StackComponent from.

return

StackComponent

The created StackComponent instance.

Properties

config

@property
def config(self) -> StackComponentConfig:
    """Returns the configuration of the stack component."""

return

StackComponentConfig

The component’s configuration object.

Subclasses should override this property to return their specific config class.

settings_class

@property
def settings_class(self) -> Optional[Type[BaseSettings]]:
    """Class specifying available settings for this component."""

return

Type[BaseSettings] | None

Optional settings class for configuring component behavior.

validator

@property
def validator(self) -> Optional[StackValidator]:
    """The optional validator of the stack component."""

return

StackValidator | None

Optional validator to check stack compatibility.

requirements

@property
def requirements(self) -> Set[str]:
    """Set of PyPI requirements for the component."""

return

Set[str]

PyPI packages required by this component.

apt_packages

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

return

List[str]

APT packages required by this component.

local_path

@property
def local_path(self) -> Optional[str]:
    """Path to a local directory for persistent information."""

return

str | None

Optional path to local storage directory.

This path must be relative to the ZenML local stores directory for proper container mounting.

log_file

@property
def log_file(self) -> Optional[str]:
    """Optional path to a log file for the stack component."""

Core Methods

get_settings

def get_settings(
    self,
    container: Union[
        Step,
        StepRunResponse,
        StepRunInfo,
        PipelineSnapshotBase,
        PipelineSnapshotResponse,
        PipelineRunResponse,
    ],
) -> BaseSettings:
    """Gets settings for this stack component."""

container

Step | StepRunInfo | PipelineSnapshot

required

The step or pipeline from which to get settings.

return

BaseSettings

Settings for this component merged with container settings.

get_connector

def get_connector(self) -> Optional[ServiceConnector]:
    """Returns the connector linked to this stack component."""

return

ServiceConnector | None

The linked service connector if configured.

connector_has_expired

def connector_has_expired(self) -> bool:
    """Checks whether the connector linked to this component has expired."""

return

bool

True if the connector has expired or isn’t linked.

Lifecycle Hooks

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.

Use this hook to set up resources before a step runs.

get_step_run_metadata

def get_step_run_metadata(
    self,
    info: StepRunInfo
) -> Dict[str, MetadataType]:
    """Get component-specific metadata after a step ran."""

info

StepRunInfo

required

Information about the step that was executed.

return

Dict[str, MetadataType]

Metadata dictionary to attach to the step run.

cleanup_step_run

def cleanup_step_run(self, info: StepRunInfo, step_failed: bool) -> None:
    """Cleans up resources after the step run is finished."""

info

StepRunInfo

required

Information about the step that was executed.

step_failed

bool

required

Whether the step failed.

cleanup

def cleanup(self) -> None:
    """Cleans up the component after it has been used."""

Pipeline-Level Hooks

get_pipeline_run_metadata

def get_pipeline_run_metadata(
    self,
    run_id: UUID
) -> Dict[str, MetadataType]:
    """Get general component-specific metadata for a pipeline run."""

run_id

UUID

required

The ID of the pipeline run.

return

Dict[str, MetadataType]

Metadata dictionary for the pipeline run.

Docker Integration

get_docker_builds

def get_docker_builds(
    self,
    snapshot: PipelineSnapshotBase
) -> List[BuildConfiguration]:
    """Gets the Docker builds required for the component."""

snapshot

PipelineSnapshotBase

required

The pipeline snapshot for which to get builds.

return

List[BuildConfiguration]

List of required Docker build configurations.

Configuration Class

StackComponentConfig

All stack components must define a configuration class inheriting from StackComponentConfig:

from zenml.stack import StackComponentConfig

class MyComponentConfig(StackComponentConfig):
    """Configuration for my custom component."""
    
    api_key: str
    endpoint: str = "https://api.example.com"
    timeout: int = 30
    
    @property
    def is_remote(self) -> bool:
        """Whether this component runs remotely."""
        return True
    
    @property
    def is_local(self) -> bool:
        """Whether this component relies on local resources."""
        return False

is_remote

bool

default:"False"

Override to return True if the component executes code remotely.

is_local

bool

default:"False"

Override to return True if the component relies on local resources.

is_valid

bool

default:"True"

Override to return False if the configuration is invalid.

Creating a Custom Component

from typing import ClassVar, Optional, Type
from zenml.stack import StackComponent, StackComponentConfig, Flavor
from zenml.enums import StackComponentType
from zenml.config.base_settings import BaseSettings

class MyComponentConfig(StackComponentConfig):
    """Custom component configuration."""
    api_endpoint: str
    api_key: str

class MyComponentSettings(BaseSettings):
    """Runtime settings for the component."""
    batch_size: int = 32
    parallel_workers: int = 4

class MyComponent(StackComponent):
    """Custom stack component implementation."""
    
    @property
    def config(self) -> MyComponentConfig:
        """Returns the component configuration."""
        return cast(MyComponentConfig, self._config)
    
    @property
    def settings_class(self) -> Optional[Type[BaseSettings]]:
        """Returns the settings class."""
        return MyComponentSettings
    
    def prepare_step_run(self, info: StepRunInfo) -> None:
        """Prepare for step execution."""
        settings = self.get_settings(info)
        # Use settings.batch_size, settings.parallel_workers
        print(f"Preparing with batch_size={settings.batch_size}")
    
    def cleanup(self) -> None:
        """Clean up resources."""
        print("Cleaning up component resources")

class MyComponentFlavor(Flavor):
    """Flavor for the custom component."""
    
    @property
    def name(self) -> str:
        return "my_component"
    
    @property
    def type(self) -> StackComponentType:
        return StackComponentType.ANNOTATOR  # Choose appropriate type
    
    @property
    def config_class(self) -> Type[StackComponentConfig]:
        return MyComponentConfig
    
    @property
    def implementation_class(self) -> Type[StackComponent]:
        return MyComponent

Secret References

Stack components support secret references for sensitive configuration:

from zenml.stack import StackComponentConfig

class SecureComponentConfig(StackComponentConfig):
    """Component with secret reference support."""
    
    # Regular configuration
    endpoint: str
    
    # Secret reference: {{secret_name.key}}
    api_key: str

# When creating the component:
# api_key can be set to "{{my_secret.api_key}}"
# At runtime, ZenML will resolve this to the actual secret value

Stack

Learn about the Stack class

BaseOrchestrator

Example of a stack component type

Stack Components Guide

Guide to all stack component types

Custom Components

How to create custom components

Core API

Context & Metadata

Artifacts

Tags & Organization

Utilities

Stack Components

​Overview

​Class Definition

​Component Types

​Initialization

​Class Methods

​from_model

​Properties

​config

​settings_class

​validator

​requirements

​apt_packages

​local_path

​log_file

​Core Methods

​get_settings

​get_connector

​connector_has_expired

​Lifecycle Hooks

​prepare_step_run

​get_step_run_metadata

​cleanup_step_run

​cleanup

​Pipeline-Level Hooks

​get_pipeline_run_metadata

​Docker Integration

​get_docker_builds

​Configuration Class

​StackComponentConfig

​Creating a Custom Component

​Secret References

​See Also

Stack

BaseOrchestrator

Stack Components Guide

Custom Components

Build docs developers (and LLMs) love

Overview

Class Definition

Component Types

Initialization

Class Methods

from_model

Properties

config

settings_class

validator

requirements

apt_packages

local_path

log_file

Core Methods

get_settings

get_connector

connector_has_expired

Lifecycle Hooks

prepare_step_run

get_step_run_metadata

cleanup_step_run

cleanup

Pipeline-Level Hooks

get_pipeline_run_metadata

Docker Integration

get_docker_builds

Configuration Class

StackComponentConfig

Creating a Custom Component

Secret References

See Also