ipeline(....):

"""Pipeline with cache disabled"""The setting above will disable caching for all steps in the pipeline unless a step explicitly sets enable_cache=True ( see below).

When writing your pipelines, be explicit. This makes it clear when looking at the code if caching is enabled or disabled for any given pipeline.

Dynamically configuring caching for a pipeline run

Sometimes you want to have control over caching at runtime instead of defaulting to the hard-coded pipeline and step decorator settings. ZenML offers a way to override all caching settings at runtime:

first_pipeline = first_pipeline.with_options(enable_cache=False)

The code above disables caching for all steps of your pipeline, no matter what you have configured in the @step or @pipeline decorators.

The with_options function allows you to configure all sorts of things this way. We will learn more about it in the coming chapters!

Caching at a step-level

Caching can also be explicitly configured at a step level via a parameter of the @step decorator:

@step(enable_cache=False)

def import_data_from_api(...):

"""Import most up-to-date data from public api"""

...

The code above turns caching off for this step only.

You can also use with_options with the step, just as in the pipeline:

import_data_from_api = import_data_from_api.with_options(enable_cache=False)

# use in your pipeline directly

Code Example

This section combines all the code from this section into one simple script that you can use to see caching easily:

from typing_extensions import Tuple, Annotated

import pandas as pd

from sklearn.datasets import load_iris

from sklearn.model_selection import train_test_split

from sklearn.base import ClassifierMixin

from sklearn.svm import SVC

from zenml import pipeline, step

from zenml.logger import get_logger

logger = get_logger(__name__)

@step

def training_data_loader() -> Tuple[

Annotated[pd.DataFrame, "X_train"],

Annotated[pd.DataFrame, "X_test"],

Annotated[pd.Series, "y_train"],

Annotated[pd.Series, "y_test"],

]: