ut_materializers=MyMaterializer)

first_pipeline()Due to the typing of the inputs and outputs and the ASSOCIATED_TYPES attribute of the materializer, you won't necessarily have to add .configure(output_materializers=MyMaterializer) to the step. It should automatically be detected. It doesn't hurt to be explicit though.

This will now work as expected and yield the following output:

Creating run for pipeline: `first_pipeline`

Cache enabled for pipeline `first_pipeline`

Using stack `default` to run pipeline `first_pipeline`...

Step `my_first_step` has started.

Step `my_first_step` has finished in 0.081s.

Step `my_second_step` has started.

The following object was passed to this step: `my_object`

Step `my_second_step` has finished in 0.048s.

Pipeline run `first_pipeline-22_Apr_22-10_58_51_135729` has finished in 0.153s.

import logging

import os

from typing import Type

from zenml import step, pipeline

from zenml.enums import ArtifactType

from zenml.materializers.base_materializer import BaseMaterializer

class MyObj:

def __init__(self, name: str):

self.name = name

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)

@step

def my_first_step() -> MyObj:

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

return MyObj("my_object")

my_first_step.configure(output_materializers=MyMaterializer)

@step

def my_second_step(my_obj: MyObj) -> None:

"""Step that log 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)