Post-Execution Workflow

After executing a pipeline, the user needs to be able to fetch it from history and perform certain tasks. This page captures these workflows at an orbital level.
Accessing past pipeline runs
In the context of a post-execution workflow, there is an implied hierarchy of some basic ZenML components:
1
repository -> pipelines -> runs -> steps -> outputs
2
​
3
# where -> implies a 1-many relationship.
Copied!
Repository
The highest level Repository object is where to start from.
1
from zenml.repository import Repository
2
​
3
repo = Repository()
Copied!
Pipelines
The repository contains a collection of all created pipelines with at least one run sorted by the time of their first run from oldest to newest.
1
# get all pipelines from all stacks
2
pipelines = repo.get_pipelines()  
3
​
4
# now you can get pipelines by index
5
pipeline_x = pipelines[-1]
6
​
7
# or get one pipeline by name and/or stack key
8
pipeline_x = repo.get_pipeline(pipeline_name=..., stack_key=...)
Copied!
The pipeline name is the name of the decorated function (or the Pipeline class in case of the Class Based API).
1
from zenml.pipelines import pipeline
2
​
3
@pipeline()
4
def this_is_the_pipeline_name():
5
    ...
Copied!
Be careful when accessing pipelines by index. Even if you just ran a pipeline it might not be at index -1, due to the fact that the pipelines are sorted by time of first run. As such it is recommended to access the pipeline by its name
Runs
Each pipeline can be executed many times. You can easily get a list of all runs like this
1
runs = pipeline_x.runs  # all runs of a pipeline chronologically ordered
2
​
3
# get the last run by index, runs are ordered by execution time in ascending order
4
run = runs[-1]
5
​
6
# or get a specific run by name
7
run = pipeline_x.get_run(run_name=...)
Copied!
Steps
Within a given pipeline run you can now zoom in further on the individual steps.
1
steps = run.steps  # all steps of a pipeline
2
step = steps[0]
3
print(step.entrypoint_name)
Copied!
The steps are ordered by time of execution. Depending on the orchestrator, steps can be run in parallel. As such, accessing steps by index can be unreliable across different runs. Instead, it makes sense to access steps by their name.
1
run.get_step(name=...)
Copied!
The step name is the name of the decorated function (or the Step class in case of the class-based API).
1
from zenml.steps import step
2
​
3
@step
4
def this_is_the_step_name():
5
    ...
Copied!
Outputs
Most of your steps will probably create outputs. You'll be able to inspect these outputs like this:
1
# The outputs of a step
2
# if there are multiple outputs they are accessible by name
3
output = step.outputs["output_name"]
4
​
5
# if one output, use the `.output` property instead 
6
output = step.output 
7
​
8
# will read the value into memory
9
output.read()  
Copied!
The names of the outputs can be found in the Output typing for your steps:
1
from zenml.steps import step, Output
2
​
3
@step
4
def some_step() -> Output(output_name=int):
5
    ...
Copied!

Developer Guide - Previous

Runtime Configuration

Next - Developer Guide

Materializers

Last modified 10d ago

Export as PDF

Copy link

Edit on GitHub

Contents

Accessing past pipeline runs