Inspect Pipeline Runs

Once a pipeline run has completed, we can access it using the ZenML Repository, about which you will find more details in a later section.
Each pipeline can have multiple runs associated with it, and for each run there might be several outputs for each step. Thus, to inspect a specific output, we first need to access the respective pipeline, then fetch the respective run, and then choose the step output of that specific run.
The overall hierarchy looks like this:
repository -> pipelines -> runs -> steps -> outputs
​
# where -> implies a 1-many relationship.
Let us investigate how to traverse this hierarchy level by level:
Repository
The highest level Repository object is where to start from.
from zenml.repository import Repository
​
​
repo = Repository()
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.
You can either access this collection via the get_pipelines() method or query a specific pipeline by name using get_pipeline(pipeline=...):
# get all pipelines from all stacks
pipelines = repo.get_pipelines()  
​
# now you can get pipelines by index
pipeline_with_latest_initial_run_time = pipelines[-1]
​
# or get one pipeline by name
pipeline_x = repo.get_pipeline(pipeline="example_pipeline")
​
# or even use the pipeline class
pipeline_x = repo.get_pipeline(pipeline=example_pipeline)
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. Instead, it is recommended to access the pipeline using the pipeline class, an instance of the class or even the name of the pipeline as a string: get_pipeline(pipeline=...).
Runs
Each pipeline can be executed many times. You can get a list of all runs using the runs attribute of a pipeline. Or, you can query a specific run by run name using the get_run(run_name=...) method:
# get all runs of a pipeline chronologically ordered
runs = pipeline_x.runs 
​
# get the last run by index, runs are ordered by execution time in ascending order
last_run = runs[-1]
​
# or get a specific run by name
run = pipeline_x.get_run(run_name=...)
Calling pipeline.runs can currently be very slow when using remote metadata stores as all run data need to be transferred from the cloud to the local machine.
Alternatively, you can also access the runs from the pipeline class/instance itself.
from zenml.pipelines import pipeline
​
# Definition of pipeline
@pipeline
def example_pipeline(...):
    ...
​
# Instantiation and execution of pipeline
pipe = example_pipeline(...)
pipe.run()
​
# get all runs of the defined pipeline chronologically ordered
runs = example_pipeline.get_runs()
​
# get all runs of the instantiated pipeline chronologically ordered
runs = pipe.get_runs()
​
# get the last run by index, runs are ordered by execution time in ascending order
last_run = runs[-1]
​
# or get a specific run by name
run = example_pipeline.get_run(run_name=...)
Each run has a collection of useful metadata which you can access:
git_sha The Git commit SHA that the pipeline run was performed on. This will only be set if the pipeline code is in a git repository and there are no uncommitted files when running the pipeline.
commit = run.git_sha
status The status of a pipeline run can also be found here. There are four possible states: failed, completed, running, cached:
status = run.status
runtime_configuration Currently the runtime configuration contains information about the schedule that was used for the run, the run_name and the path to the file containing the pipeline.
runtime_config = run.runtime_configuration
Steps
Within a given pipeline run you can now further zoom in on individual steps using the steps attribute or by querying a specific step using the get_step(step=...) method.
# get all steps of a pipeline for a given run
steps = run.steps
​
# get the step that was executed first
first_step = steps[0]
​
# or get a specific step by name
step = run.get_step(step="first_step")
The step name refers to the pipeline attribute and not the class name of the steps that implement the step for a pipeline instance.
# Definition of pipeline
@pipeline
def example_pipeline(step_1, step_2):
    ...
​
# Initialize a new pipeline run
pipe = example_pipeline(step_1=first_step(), step_2=second_step())
pipe.run()
​
# Get the first step
pipe.get_runs()[-1].get_step(step="step_1")
​
# This won't work:
# pipe.get_runs()[-1].get_step(step="first_step")
The steps are ordered by time of execution. Depending on the orchestrator, steps can be run in parallel. Thus, accessing steps by index can be unreliable across different runs, and it is recommended to access steps by the step class, an instance of the class or even the name of the step as a string: get_step(step=...) instead.
Outputs
Finally, this is how you can inspect the output of a step:
If there only is a single output, use the output attribute
If there are multiple outputs, use the outputs attribute, which is a dictionary that can be indexed using the name of an output:
# The outputs of a step
# if there are multiple outputs they are accessible by name
output = step.outputs["output_name"]
​
# if there is only one output, use the `.output` property instead 
output = step.output 
​
# read the value into memory
output.read()  
The names of the outputs can be found in the Output typing of your steps:
from zenml.steps import step, Output
​
​
@step
def some_step() -> Output(output_name=int):
    ...
Code Example
Putting it all together, this is how we can access the output of the last step of our example pipeline from the previous sections:
from zenml.repository import Repository
​
repo = Repository()
pipeline = repo.get_pipeline(pipeline="first_pipeline")
last_run = pipeline.runs[-1]
last_step = last_run.steps[-1]
model = last_step.output.read()
or alternatively:
# Definition of pipeline
@pipeline
def example_pipeline(step_1, step_2):
    ...
# Initialize a new pipeline run
pipe = example_pipeline(step_1=first_step(), step_2=second_step())
pipe.run()
​
# Get the first step
step_1 = pipe.get_runs()[-1].get_step(step="step_1")
output = step_1.output.read()

Choose Functional vs. Class-Based API

Configure Automated Caching

Last modified 7d ago

Copy link

Outline