Post-Execution Workflow
Inspect a Finished Pipeline Run.
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!
Export as PDF
Copy link
Edit on GitHub