_init__(self, name: str):

self.name = name

@stepdef my_first_step() -> MyObj:

"""Step that returns an object of type MyObj."""

return MyObj("my_object")

@step

def my_second_step(my_obj: MyObj) -> None:

"""Step that logs the input object and returns nothing."""

logging.info(

f"The following object was passed to this step: `{my_obj.name}`"

@pipeline

def first_pipeline():

output_1 = my_first_step()

my_second_step(output_1)

first_pipeline()

Running the above without a custom materializer will work but print the following warning:

No materializer is registered for type MyObj, so the default Pickle materializer was used. Pickle is not production ready and should only be used for prototyping as the artifacts cannot be loaded when running with a different Python version. Please consider implementing a custom materializer for type MyObj

To get rid of this warning and make our pipeline more robust, we will subclass the BaseMaterializer class, listing MyObj in ASSOCIATED_TYPES, and overwriting load() and save():

import os

from typing import Type

from zenml.enums import ArtifactType

from zenml.materializers.base_materializer import BaseMaterializer

class MyMaterializer(BaseMaterializer):

ASSOCIATED_TYPES = (MyObj,)

ASSOCIATED_ARTIFACT_TYPE = ArtifactType.DATA

def load(self, data_type: Type[MyObj]) -> MyObj:

"""Read from artifact store."""

with self.artifact_store.open(os.path.join(self.uri, 'data.txt'), 'r') as f:

name = f.read()

return MyObj(name=name)

def save(self, my_obj: MyObj) -> None:

"""Write to artifact store."""

with self.artifact_store.open(os.path.join(self.uri, 'data.txt'), 'w') as f:

f.write(my_obj.name)

Pro-tip: Use the self.artifact_store property to ensure your materialization logic works across artifact stores (local and remote like S3 buckets).

Now, ZenML can use this materializer to handle the outputs and inputs of your customs object. Edit the pipeline as follows to see this in action:

my_first_step.configure(output_materializers=MyMaterializer)

first_pipeline()