Migration guide 0.39.1 → 0.41.0

How to migrate your ZenML pipelines and steps from version <=0.39.1 to 0.41.0.

ZenML versions 0.40.0 to 0.41.0 introduced a new and more flexible syntax to define ZenML steps and pipelines. This page contains code samples that show you how to upgrade your steps and pipelines to the new syntax.

Newer versions of ZenML still work with pipelines and steps defined using the old syntax, but the old syntax is deprecated and will be removed in the future.

Overview

from typing import Optional

from zenml.steps import BaseParameters, Output, StepContext, step

from zenml.pipelines import pipeline

# Define a Step

class MyStepParameters(BaseParameters):

param_1: int

param_2: Optional[float] = None

@step

def my_step(

params: MyStepParameters, context: StepContext,

) -> Output(int_output=int, str_output=str):

result = int(params.param_1 * (params.param_2 or 1))

result_uri = context.get_output_artifact_uri()

return result, result_uri

# Run the Step separately

my_step.entrypoint()

# Define a Pipeline

@pipeline

def my_pipeline(my_step):

my_step()

step_instance = my_step(params=MyStepParameters(param_1=17))

pipeline_instance = my_pipeline(my_step=step_instance)

# Configure and run the Pipeline

pipeline_instance.configure(enable_cache=False)

schedule = Schedule(...)

pipeline_instance.run(schedule=schedule)

# Fetch the Pipeline Run

last_run = pipeline_instance.get_runs()[0]

int_output = last_run.get_step["my_step"].outputs["int_output"].read()

from typing import Annotated, Optional, Tuple

from zenml import get_step_context, pipeline, step

from zenml.client import Client

# Define a Step

@step

def my_step(

param_1: int, param_2: Optional[float] = None

) -> Tuple[Annotated[int, "int_output"], Annotated[str, "str_output"]]:

result = int(param_1 * (param_2 or 1))

result_uri = get_step_context().get_output_artifact_uri()

return result, result_uri

# Run the Step separately

my_step()

# Define a Pipeline

@pipeline