ifact):  # rather than pd.DataFrame

pass

ExampleThe following shows an example of how unmaterialized artifacts can be used in the steps of a pipeline. The pipeline we define will look like this:

s1 -> s3

s2 -> s4

from typing_extensions import Annotated  # or `from typing import Annotated on Python 3.9+

from typing import Dict, List, Tuple

from zenml.artifacts.unmaterialized_artifact import UnmaterializedArtifact

from zenml import pipeline, step

@step

def step_1() -> Tuple[

Annotated[Dict[str, str], "dict_"],

Annotated[List[str], "list_"],

]:

return {"some": "data"}, []

@step

def step_2() -> Tuple[

Annotated[Dict[str, str], "dict_"],

Annotated[List[str], "list_"],

]:

return {"some": "data"}, []

@step

def step_3(dict_: Dict, list_: List) -> None:

assert isinstance(dict_, dict)

assert isinstance(list_, list)

@step

def step_4(

dict_: UnmaterializedArtifact,

list_: UnmaterializedArtifact,

) -> None:

print(dict_.uri)

print(list_.uri)

@pipeline

def example_pipeline():

step_3(*step_1())

step_4(*step_2())

example_pipeline()

Interaction with custom artifact stores

When creating a custom artifact store, you may encounter a situation where the default materializers do not function properly. Specifically, the self.artifact_store.open method used in these materializers may not be compatible with your custom store due to not being implemented properly.

In this case, you can create a modified version of the failing materializer by copying it and modifying it to copy the artifact to a local path, then opening it from there. For example, consider the following implementation of a custom PandasMaterializer that works with a custom artifact store. In this implementation, we copy the artifact to a local path because we want to use the pandas.read_csv method to read it. If we were to use the self.artifact_store.open method instead, we would not need to make this copy.