ZenML Materializers 完全指南从内置类型到自定义数据类型的序列化、存储与可视化【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenmlMaterializers物化器是 ZenML 构件Artifact体系中的核心机制负责定义某一种数据类型如何在 ML Pipeline 中被序列化、写入构件存储Artifact Store、读取还原、可视化展示与元数据提取。本文以 ZenML 官方文档中 materializers 指南为主体结合仓库源码src/zenml/materializers/与配置实现系统讲解内置 materializer、Path文件/目录透传、自定义 materializer 的完整开发流程以及跳过物化的UnmaterializedArtifact高级用法。读完本文你将掌握让任意自定义数据类型在 ZenML Pipeline 各步骤间可靠传递、并在 Dashboard 中可视化的完整实战方案。什么是 MaterializerMaterializer 是一个定义了某一种数据类型如何在 Pipeline 中流转的类它统一负责六类职责序列化Serialized将 Python 对象转换为可存储的格式保存Saved将序列化结果写入构件存储加载Loaded从构件存储读取数据反序列化Deserialized将存储内容还原为 Python 对象可视化Visualized在 ZenML Dashboard 中展示构件内容分析Analyzed提取元数据用于追踪与检索。Materializer 是 Python 代码与底层存储系统之间的桥梁。当步骤 A 输出一个对象、步骤 B 将其作为输入时ZenML 会在步骤 A 结束时调用 materializer 的save()把对象写入构件存储并在步骤 B 开始时调用load()将其还原从而保证任意数据类型都能被正确保存、加载与可视化。从源码看所有 materializer 都继承自BaseMaterializer。基类通过元类BaseMaterializerMeta在类定义时自动校验并注册到全局materializer_registry子类必须声明ASSOCIATED_TYPES至少一种关联类型否则会抛出MaterializerInterfaceErrorASSOCIATED_ARTIFACT_TYPE必须为合法枚举值。基类实例化时接收两个关键参数uri构件数据在存储中的路径与artifact_store构件存储句柄artifact_store缺省时会回退到当前激活 Stack 的构件存储。内置 Materializer核心 MaterializerZenML 为常见 Python 数据类型内置了开箱即用的 materializerMaterializer处理的数据类型存储格式BuiltInMaterializerbool、float、int、str、None.jsonBytesMaterializerbytes.txtBuiltInContainerMaterializerdict、list、set、tuple目录NumpyMaterializernp.ndarray.npyPandasMaterializerpd.DataFrame、pd.Series.csv若安装了parquet则为.gzipPydanticMaterializerpydantic.BaseModel.jsonDataclassMaterializer可 JSON 序列化的 Pythondataclass.jsonServiceMaterializerzenml.services.service.BaseService.jsonStructuredStringMaterializerzenml.types.CSVString、HTMLString、MarkdownString.csv/.html/.mdPathMaterializerpathlib.Path.tar.gz目录或直接拷贝文件这些核心实现可以在仓库src/zenml/materializers/中直接查看BuiltInMaterializer与BuiltInContainerMaterializer定义于 built_in_materializer.py。其中容器类型非常智能如果整个容器可 JSON 序列化就整体写入单个data.json否则会把每个元素物化到独立子目录并通过metadata.json记录每个元素路径、类型与所用 materializer加载时逐元素还原set、tuple会先转为list再处理。BuiltInContainerMaterializer还实现了get_item_count与load_item支持对list/tuple类构件按索引按需加载。BytesMaterializer以二进制方式直接读写data.txt避免bytes无法 JSON 序列化的问题。PydanticMaterializer定义于 pydantic_materializer.py序列化时使用model_dump(modejson)写入文件。兼容性提示当前版本 ZenML 创建的 Pydantic 构件存储于data_v2.json而 ZenML 0.94.2创建的旧构件以data.json存储当时存在双重编码。PydanticMaterializer.load()会先探测data_v2.json不存在时回退到data.json并用model_validate_json兼容解码因此升级后既有运行记录仍然可读。此外ZenML 还提供CloudpickleMaterializer可对任意对象用 cloudpickle 可见它在save()时会额外写入python_version.txtload()时若检测到 Python 版本不一致会发出警告。正因其不可靠该类设置了SKIP_REGISTRATION True仅作为 materializer 注册表中的最后兜底见下文全局注册与查找机制。生产环境中请为你的专属数据类型实现自定义 materializer。Dataclass 构件DataclassMaterializer允许直接传递可 JSON 序列化的 Python dataclass无需编写任何自定义 materializerfrom dataclasses import dataclass from zenml import step dataclass class TrainingConfig: learning_rate: float epochs: int step def make_config() - TrainingConfig: return TrainingConfig(learning_rate0.01, epochs10)这适用于 Pydantic 能够序列化为 JSON 的 dataclass。如果 dataclass 内包含打开的文件句柄、存活的模型对象、数据库连接或其他任意 Python 对象则应改用自定义 materializer。从注册表实现看MaterializerRegistry.__getitem__在常规类型查找失败后会尝试用DataclassMaterializer.can_save_type(key)判断该类型是否为可处理的 dataclass命中则自动使用DataclassMaterializer见 materializer_registry.py。在步骤之间传递文件与目录PathMaterializer让你可以在步骤之间传递pathlib.Path对象特别适合数据集目录、导出的模型文件等基于文件系统的构件。其行为如下步骤返回Path指向目录时目录会被压缩为.tar.gz归档并上传到构件存储步骤返回Path指向单个文件时文件被直接拷贝到构件存储下游步骤接收该Path时materializer 会把内容下载到本地临时目录并返回指向它的Path。from pathlib import Path from typing import Annotated from zenml import step, pipeline step def prepare_dataset(num_samples: int 100) - Annotated[Path, dataset_dir]: Prepare a dataset directory with training files. output_dir Path(training_data) output_dir.mkdir(exist_okTrue) # Write training files into the directory (output_dir / features.csv).write_text(feature1,feature2\n1.0,2.0\n) (output_dir / labels.csv).write_text(label\n1\n) # ZenML will tar.gz this directory and upload it to the artifact store return output_dir step def train_model(dataset_dir: Path) - None: Train a model using the dataset directory. # dataset_dir points to a local temp directory with the extracted contents features (dataset_dir / features.csv).read_text() labels (dataset_dir / labels.csv).read_text() print(fTraining with features: {features}) pipeline def training_pipeline(): dataset prepare_dataset() train_model(dataset)该机制在远程编排器Kubernetes、Vertex AI 等下同样透明可用——每个步骤运行在不同 Pod 上时构件存储充当了共享传输层。从源码看PathMaterializer的存储文件命名约定为目录归档data.tar.gz、单文件file_data。save()使用shutil.make_archive(formatgztar, root_dir...)生成相对路径归档load()在get_temporary_directory(delete_at_exitFalse)创建的临时目录中解包该目录会在步骤执行结束后自动清理。尤其值得关注的是解包前会对每个 tar 成员调用_is_safe_tar_member校验防止路径穿越类安全攻击见 path_materializer.py。兼容性开关如果你更偏好旧行为——Path对象仅用 cloudpickle 序列化路径字符串不保留文件内容——可通过环境变量ZENML_DISABLE_PATH_MATERIALIZERtrue禁用PathMaterializer。该变量在 path_materializer.py 中直接控制SKIP_REGISTRATION即设置后PathMaterializer不会注册到全局注册表。集成专属 Materializer安装 ZenML 集成integration后会额外提供大量针对机器学习生态的 materializer集成Materializer处理的数据类型存储格式bentomlBentoMaterializerbentoml.Bento.bentodeepchecksDeepchecksResultMateriailzerdeepchecks.CheckResult、SuiteResult.jsonevidentlyEvidentlyProfileMaterializerevidently.Profile.jsongreat_expectationsGreatExpectationsMaterializerExpectationSuite、CheckpointResult.jsonhuggingfaceHFDatasetMaterializerdatasets.Dataset、DatasetDict目录huggingfaceHFPTModelMaterializertransformers.PreTrainedModel目录huggingfaceHFTFModelMaterializertransformers.TFPreTrainedModel目录huggingfaceHFTokenizerMaterializertransformers.PreTrainedTokenizerBase目录lightgbmLightGBMBoosterMaterializerlgbm.Booster.txtlightgbmLightGBMDatasetMaterializerlgbm.Dataset.binaryneural_prophetNeuralProphetMaterializerNeuralProphet.ptpillowPillowImageMaterializerPillow.Image.PNGpolarsPolarsMaterializerpl.DataFrame、pl.Series.parquetpycaretPyCaretMaterializer任意sklearn、xgboost、lightgbm或catboost模型.pklpytorchPyTorchDataLoaderMaterializertorch.Dataset、torch.DataLoader.ptpytorchPyTorchModuleMaterializertorch.Module.ptscipySparseMaterializerscipy.spmatrix.npzsparkSparkDataFrameMaterializerpyspark.DataFrame.parquetsparkSparkModelMaterializerpyspark.Transformer/pyspark.Estimator.parquettensorflowKerasMaterializertf.keras.Model目录tensorflowTensorflowDatasetMaterializertf.Dataset目录whylogsWhylogsMaterializerwhylogs.DatasetProfileView.pbxgboostXgboostBoosterMaterializerxgb.Booster.jsonxgboostXgboostDMatrixMaterializerxgb.DMatrix.binaryjaxJAXArrayMaterializerjax.Array.npymlxMLXArrayMaterializermlx.core.array.npy注意使用基于 Docker 的编排器时必须在DockerSettings中指定相应集成确保容器内存在对应的 materializer 代码。值得注意的是一处可观测的工程细节Pandas 相关的 materializer 已从核心包迁入集成src/zenml/materializers/pandas_materializer.py现在只是一个兼容占位文件会引导你zenml integration install pandas并从zenml.integrations.pandas.materializers导入旧构件版本记录的 materializer 源码位置仍可解析因此无需数据库迁移。这提醒我们materializer 的源码路径被持久化在构件版本记录中重构时应保持向后兼容。创建自定义 Materializer当内置 materializer 无法覆盖你的自定义数据类型时创建自定义 materializer 通常只需四步。1. 定义 Materializer 类创建一个继承自BaseMaterializer的新类import os import json from typing import Type, Any, Dict from zenml.materializers.base_materializer import BaseMaterializer from zenml.enums import ArtifactType, VisualizationType from zenml.metadata.metadata_types import MetadataType # Assume MyClass is your custom class defined elsewhere # from mymodule import MyClass class MyClassMaterializer(BaseMaterializer): Materializer for MyClass objects. # List the data types this materializer can handle ASSOCIATED_TYPES (MyClass,) # Define what type of artifact this is (usually DATA or MODEL) ASSOCIATED_ARTIFACT_TYPE ArtifactType.DATA def load(self, data_type: Type[Any]) - MyClass: Load MyClass from storage. # Implementation here filepath os.path.join(self.uri, data.json) with self.artifact_store.open(filepath, r) as f: data json.load(f) # Create and return an instance of MyClass return MyClass(**data) def save(self, data: MyClass) - None: Save MyClass to storage. # Implementation here filepath os.path.join(self.uri, data.json) with self.artifact_store.open(filepath, w) as f: json.dump(data.to_dict(), f) def save_visualizations(self, data: MyClass) - Dict[str, VisualizationType]: Generate visualizations for the dashboard. # Optional - generate visualizations vis_path os.path.join(self.uri, visualization.html) with self.artifact_store.open(vis_path, w) as f: f.write(data.to_html()) return {vis_path: VisualizationType.HTML} def extract_metadata(self, data: MyClass) - Dict[str, MetadataType]: Extract metadata for tracking. # Optional - extract metadata return { name: data.name, created_at: data.created_at, num_records: len(data.records) }定义类时注意只要在类体声明了ASSOCIATED_TYPESBaseMaterializerMeta元类就会在导入时自动把该 materializer 注册进全局注册表materializer_registry.register_materializer_type因此无需额外手动注册见 base_materializer.py。2. 在 Pipeline 中使用自定义 Materializer定义完成后通过output_materializers参数在步骤上指定from zenml import step, pipeline # from mymodule import MyClass, MyClassMaterializer step(output_materializersMyClassMaterializer) def create_my_class() - MyClass: Create an instance of MyClass. return MyClass(nametest, records[1, 2, 3]) step def use_my_class(my_obj: MyClass) - None: Use the MyClass instance. print(fName: {my_obj.name}, Records: {my_obj.records}) pipeline def custom_pipeline(): data create_my_class() use_my_class(data)output_materializers参数由BaseStep提供支持单个类、类列表或按输出名映射的字典。在内部MaterializerRegistry.__getitem__会沿着类型的 MRO方法解析顺序逐级查找已注册的 materializer找到最近的父类对应实现即命中查找失败则回退到默认的CloudpickleMaterializer见 materializer_registry.py。3. 多输出使用不同 Materializer当一个步骤有多个输出、且需要不同 materializer 时使用字典形式按输出名映射from typing import Tuple, Annotated step(output_materializers{ obj1: MyClass1Materializer, obj2: MyClass2Materializer }) def create_objects() - Tuple[ Annotated[MyClass1, obj1], Annotated[MyClass2, obj2] ]: Create instances of different classes. return MyClass1(), MyClass2()4. 全局注册 Materializer你可以将 materializer 全局注册覆盖某种类型的默认实现from zenml.materializers.materializer_registry import materializer_registry from zenml.materializers.base_materializer import BaseMaterializer import pandas as pd # Create a custom pandas materializer class FastPandasMaterializer(BaseMaterializer): # Implementation here ... # Register it for pandas DataFrames globally materializer_registry.register_and_overwrite_type( keypd.DataFrame, type_FastPandasMaterializer )这里register_and_overwrite_type与普通register_materializer_type的关键区别在于后者遇到同名键会跳过保留先注册者而前者会无条件覆盖见 materializer_registry.py。全局注册的查找语义不变——按key.__mro__逐级匹配父类。Materializer 实现细节实现自定义 materializer 时需要关注以下几个核心方面。存储处理self.uri属性指向构件在存储中应存放的目录路径用它来创建数据文件或子目录。读写文件时始终使用self.artifact_store.open()而非直接的文件 I/O以保证在不同构件存储本地文件系统、S3、GCS 等上的一致行为。artifact_store属性来自构造参数缺省时回退到当前激活 Stack 的构件存储见 base_materializer.py。另外值得了解的是内容哈希与版本校验机制基类在load()前会设置expected_content_hashmaterializer 可实现compute_content_hash()在保存时计算哈希。BuiltInMaterializer、BuiltInContainerMaterializer、PydanticMaterializer均实现了该方法如 built_in_materializer.py 用 MD5 对序列化 JSON 计算CloudpickleMaterializer则在load()时校验 SHA-256 防止构件内容被篡改。可视化支持save_visualizations()方法允许你创建将在 ZenML Dashboard 中展示的可视化可同时返回多种类型VisualizationType.HTML内嵌 HTML 内容VisualizationType.MARKDOWNMarkdown 内容VisualizationType.IMAGE图片文件VisualizationType.CSVCSV 表格基类 docstring 给出的参考实现会把可视化文件写入self.uri下再以{可视化路径: 类型}字典返回见 base_materializer.py。内置 materializer 也大量使用了该能力BuiltInMaterializer把data.json注册为 JSON 可视化、BytesMaterializer注册为 MARKDOWN、PydanticMaterializer同样注册 JSON 可视化。通过环境变量配置可视化部分 materializer 支持用环境变量定制可视化行为。例如ZENML_PANDAS_SAMPLE_ROWS控制PandasMaterializer生成样本可视化时展示的行数默认 10 行。该默认值定义于 pandas_materializer.py并在生成可视化时通过os.environ.get(ZENML_PANDAS_SAMPLE_ROWS, DEFAULT_SAMPLE_ROWS)读取。元数据提取extract_metadata()方法用于提取构件的关键信息供索引与检索使用这些元数据会与构件一同展示在 Dashboard 中。基类的extract_full_metadata()会合并基础元数据与自定义元数据其中基础元数据自动包含storage_size构件存储占用字节数递归计算目录内所有文件大小见 base_materializer.py。内置实现各有特色BuiltInMaterializer对数值类型记录string_representationBuiltInContainerMaterializer记录lengthPydanticMaterializer记录完整的 JSON Schema。临时文件处理构件时如果需要临时目录使用get_temporary_directory()辅助方法with self.get_temporary_directory() as temp_dir: # Process files in the temporary directory # Files will be automatically cleaned up该方法基于tempfile.mkdtemp(prefixzenml-)创建目录见 base_materializer.py。两个参数控制生命周期delete_at_exitTrue时上下文退出即删除delete_at_exitFalse但delete_after_step_executionTrue时目录会注册到当前步骤上下文的清理回调中在步骤执行完毕后删除。若在步骤执行上下文之外调用且未请求退出时删除则需自行负责清理。完整示例一个端到端的自定义 Materializer以下是一个完整可运行的自定义 materializer 示例import os import json from typing import Type, Any, Dict from zenml.materializers.base_materializer import BaseMaterializer from zenml.enums import ArtifactType class MyObj: def __init__(self, name: str): self.name name def to_dict(self): return {name: self.name} classmethod def from_dict(cls, data): return cls(namedata[name]) class MyMaterializer(BaseMaterializer): Materializer for MyObj objects. ASSOCIATED_TYPES (MyObj,) ASSOCIATED_ARTIFACT_TYPE ArtifactType.DATA def load(self, data_type: Type[Any]) - MyObj: Load MyObj from storage. filepath os.path.join(self.uri, data.json) with self.artifact_store.open(filepath, r) as f: data json.load(f) return MyObj.from_dict(data) def save(self, data: MyObj) - None: Save MyObj to storage. filepath os.path.join(self.uri, data.json) with self.artifact_store.open(filepath, w) as f: json.dump(data.to_dict(), f) # Usage in a pipeline step(output_materializersMyMaterializer) def create_my_obj() - MyObj: return MyObj(namemy_object) step def use_my_obj(my_obj: MyObj) - None: print(fObject name: {my_obj.name}) pipeline def my_pipeline(): obj create_my_obj() use_my_obj(obj)使用未物化的构件Unmaterialized Artifacts一般情况下构件从上游步骤输出到下游步骤输入时会由对应数据类型的 materializer 完成先序列化写入、再反序列化读取的完整流程。但有些场景你不希望物化构件而是想直接使用构件引用本身——例如需要获知构件在存储中的确切路径时。警告跳过物化可能对依赖已物化构件的下游任务产生意外影响。只有在别无选择时才跳过物化。如何跳过物化未物化的构件是zenml.materializers.UnmaterializedArtifact。它继承自ArtifactVersionResponse见 unmaterialized_artifact.py其中uri属性指向构件在存储中持久化的唯一路径。将步骤输入类型声明为UnmaterializedArtifact即可使用from zenml.artifacts.unmaterialized_artifact import UnmaterializedArtifact from zenml import step step def my_step(my_artifact: UnmaterializedArtifact): # rather than pd.DataFrame pass下面是一个更完整的例子。定义如下 Pipelines1 - s3 s2 - s4s1与s2产出完全相同的构件但s3消费物化后的构件而s4消费未物化构件——s4可以直接使用dict_.uri与list_.uri路径而不必经过物化from typing import Annotated 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()你还可以在从其他 Pipeline 触发 Snapshot 的高级用法中看到另一个使用UnmaterializedArtifact的实例。最佳实践使用 materializer 时遵循以下准则优先使用结构化格式如 JSON、CSV而非 pickle 或其他二进制格式以获得更好的跨环境兼容性。在不同构件存储上测试你的 materializer本地、S3 等确保行为一致——请始终通过self.artifact_store.open()读写这正是其跨存储一致性的来源。考虑版本化如果数据结构会随时间变化为存储格式设计版本字段并参考PydanticMaterializer对新旧文件名data_v2.json/data.json的兼容回退策略。创建可视化帮助用户在 Dashboard 中理解构件内容。提取有价值的元数据让构件更易被检索和理解。显式指定 materializer即便 ZenML 可以自动检测明确指定也让 Pipeline 更清晰、更可维护。生产环境避免使用CloudpickleMaterializer它在不同 Python 版本间不可靠。若看到CloudpickleMaterializer的警告日志说明你的数据类型没有注册专属 materializer请按本文指引实现一个。总结Materializer 是 ZenML 构件系统中极具威力的部分它让任意数据类型都能被正确存储与处理。理解内置 materializer 的职责划分基础类型、容器、Pydantic、dataclass、Path 乃至各 ML 框架集成掌握BaseMaterializer的save/load/save_visualizations/extract_metadata生命周期并通过output_materializers或全局注册表接入自定义类型你就能确保 ML Pipeline 稳健、高效地支撑工作流所需的任何数据类型——无论是配置文件、数据集目录、训练好的模型还是业务自定义对象。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考