如何构建一个端到端的机器学习流水线

我们在社区中经常听到的一个最普遍的抱怨是，手动完成机器学习工作流程（包括构建和部署模型）的成本高昂且容易出错。他们手动运行脚本对训练数据进行预处理，重新运行部署脚本，手动调整模型，并花费大量工作时间来维护先前开发的模型。

构建端到端的机器学习流水线可以让机器学习工程师只需要构建一次，然后多次重复运行和复用。这让他们能够更多地专注于部署新模型，而不是维护现有模型。它提供了专注于新模型而不是过多努力维护现有模型的能力。

当然，如果一切顺利的话 😉

在本文中，您将会：

• 1
  探索机器学习流水线的体系结构，包括其中的组件。
• 2
  学习机器学习工程师可以遵循的关键步骤和最佳实践，以构建稳健、可扩展的端到端机器学习流水线。
• 3
  使用Kubeflow Pipelines在AWS上快速构建和部署端到端的机器学习流水线。
• 4
  学习构建端到端机器学习流水线的挑战和最佳实践。

什么是机器学习流水线？

机器学习流水线由一系列链接的组件或步骤组成，定义了解决特定问题的机器学习工作流程。流水线可以让您编排自动化的机器学习工作流程步骤。这里的编排意味着工作流程步骤之间的依赖关系和数据流必须按正确的顺序完成。

您可以构建一个流水线来：

• 实现工作流程的可复现性（在类似的输入上多次运行流水线将提供类似的输出）。
• 简化机器学习工作流程中多个步骤的端到端编排，减少对机器学习团队的干预和手动操作（自动化）。
• 缩短数据和模型从实验阶段到生产阶段所需的时间。
• 使用模块化组件，使团队更多地专注于开发新解决方案，而不是维护现有解决方案，并为工作流程提供自动化。
• 轻松地复用组件（机器学习工作流程中的特定步骤），以创建和部署与外部系统集成的端到端解决方案，而无需每次重新构建。

机器学习流水线与机器学习平台

机器学习流水线是更广泛的机器学习平台的一部分。它用于在机器学习平台中简化、编排和自动化机器学习工作流程。

流水线和平台是MLOps中相关的概念，但它们指的是机器学习工作流程的不同方面。机器学习平台是为您的机器学习/人工智能团队标准化技术栈并提供工具、库和基础设施来开发、部署和运营机器学习应用程序的环境。

该平台通常包括用于机器学习生态系统的组件，如数据管理、特征存储、实验跟踪器、模型注册表、测试环境、模型服务和模型管理。它旨在为数据科学家和机器学习工程师提供统一和集成的环境，以开发和部署模型、管理数据并简化机器学习工作流程。

机器学习流水线的架构

机器学习流水线的架构可以是实时（在线）或批处理（离线）构建，具体取决于用例和生产要求。为了在本文中保持概念简单，您将了解一个典型的流水线是什么样子，而不涉及实时或批处理构建的细节。

Semi Koen的文章详细介绍了机器学习流水线的架构。

常见类型的机器学习流水线

根据ML工作流程的阶段（数据、模型和生产），ML管道包括解决不同工作流程阶段的三个不同管道。它们包括：

• 1
  数据（或输入）管道。
• 2
  模型（或训练）管道。
• 3
  服务（或生产）管道。

在大型组织中，由于功能和规模的原因，可能会有两个或多个团队负责每个管道。这些管道是互操作的，用于构建一个工作系统：

数据（输入）管道（数据采集和特征管理步骤）

该管道将原始数据从一个位置传输到另一个位置。它涵盖了整个数据移动过程，从数据采集的地方（例如通过数据流或批处理）到下游应用程序，如数据湖或机器学习模型。

模型训练管道

该管道在训练数据上训练一个或多个模型，并使用预设的超参数对其进行评估、微调并打包最佳模型，然后将其发送到诸如模型注册表或服务管道之类的应用程序。

服务管道

该管道将模型部署为生产中的预测（或评分）服务，并使用另一个服务来实现性能监控。

本文将不同的管道分类为“机器学习管道”，因为它们根据其在工作流程中的功能支持ML应用程序。此外，它们是互操作的，可以在维护期间（重新训练和持续测试）支持生产应用程序。

机器学习管道的组成要素

某些管道将通过以下三个元素为这些组件提供高级抽象：

1. 转换器：一种能够将一个数据集转换为另一个数据集的算法。
2. 估计器：在数据集上训练的算法，用于生成转换器。
3. 评估器：用于检查训练模型的准确性。

机器学习管道的组件

管道组件是机器学习工作流程中的一步，通过接受输入、处理它并生成输出来执行特定任务。这些组件包括可自动化步骤的手动工作流程过程的实现，包括：

• 数据摄取（提取和版本控制）。
• 数据验证（编写测试以检查数据质量）。
• 数据预处理。
• 模型训练和调优，根据一组选择的算法进行探索和一系列超参数进行实验。
• 模型性能分析和评估。
• 模型打包和注册。
• 模型部署。
• 模型评分。
• 模型性能监控。

使用大多数工具，管道组件将包含可执行代码，可以将其容器化（以消除依赖问题）。可以使用诸如Kubeflow Pipelines、Metaflow或ZenML之类的编排工具来管理每个步骤。

让我们简要地介绍一下每个组件。

数据摄取、提取和版本控制

该组件从机器学习管道外部的数据源摄取数据作为输入。然后，它将数据集转换为下一个步骤中将使用的格式（例如CSV、Parquet等）。在此步骤中，原始数据和版本化数据也会进行转换，以便更容易追踪其血统。

数据验证

该步骤收集转换后的数据作为输入，并通过一系列测试和验证器确保其符合下一个组件的标准。它检查数据质量问题并检测离群值和异常值。该组件还检查数据漂移或潜在的训练-服务偏差迹象，以将日志发送到其他组件或向负责的数据科学家发出警报。

如果验证测试通过，数据将被发送到下一个组件；如果测试未通过，则会记录错误并停止执行。

数据预处理和特征工程

数据清理、分离和特征工程步骤以先前组件的验证和转换后的数据作为输入。此步骤涉及的过程取决于您要解决的问题和数据。此处的过程可能包括：

• 特征选择：选择要进行清理和工程处理的最合适的特征。
• 特征清理：通过基于代码实现的上限/下限将缺失的特征值进行处理并删除离群值。
• 特征转换：转换数据中的倾斜特征（如果适用）。
• 特征创建：从现有特征创建新特征或将不同特征组合成一个新特征。
• 数据分离：将数据分成训练、测试和验证集。
• 特征标准化/归一化：将特征值转换为类似的尺度和分布值。
• 将特征发布到特征存储中，以供整个组织进行培训和推理。

再次强调，此组件中的操作是根据数据科学家的初始（手动）数据准备过程、问题和使用的数据而定的。

模型训练和调优

此组件可以从特征存储中检索准备好的特征，或从前一个组件获取准备好的数据集（训练集和验证集）作为输入。

此组件使用一系列预设的超参数来训练模型（使用网格搜索 CV、神经架构搜索或其他技术）。它还可以并行训练多个具有不同超参数值集的模型。训练好的模型作为工件发送给下一个组件。

模型评估

训练好的模型是此组件的输入，将在验证集上进行评估。您可以根据诸如 ROC、AUC、精确度、召回率和准确度等指标分析每个模型的结果。指标通常根据问题设置。然后将这些指标记录下来以供将来分析。

模型分析和验证

此组件：

• 1
  评估模型对未见数据的泛化能力。
• 2
  分析模型的可解释性，以帮助您了解计划部署的模型或模型的质量和偏差。它检查模型在数据切片上的表现以及模型的特征重要性。它是一个黑盒模型，还是可以解释其决策的？

如果您训练了多个模型，该组件还可以在测试集上评估每个模型，并提供选择最佳模型的选项。

在这里，组件还将返回帮助您了解模型是否适合目标部署环境的统计数据和元数据。例如：

• 它是否过大，无法满足基础设施要求？
• 进行预测需要多长时间？
• 进行预测时消耗多少资源（CPU 使用率、内存等）？

如果您的流水线正在部署中，该组件还可以帮助您比较训练好的模型的指标与生产中的指标是否显著降低，并在有显著降低时发出警报。

模型打包和注册

该组件将您的模型打包，以便部署到暂存或生产环境。模型工件和必要的配置文件将被打包、版本化并发送到模型注册表。

容器是一种有用的模型打包技术。它们将部署的模型封装为一个独立的评分服务，可以在任何地方运行。其他部署选项也可用，例如使用生产环境的语言重写部署代码。对于机器学习流水线，使用容器是最常见的方法。

模型部署

您可以将打包和注册的模型部署到暂存环境（作为传统软件与 DevOps）或生产环境。暂存环境用于集成测试。暂存环境是第一个类似生产环境的环境，可以与整个系统中的其他服务一起测试模型，这些服务支持应用程序的运行。例如，部署推荐服务并将其与将客户请求路由到服务的后端服务器进行测试。

一些组织可能选择在 Kubernetes 等容器编排平台上进行暂存。这取决于您用于流水线编排的工具。

虽然不推荐，但您也可以直接将打包和注册的模型部署到生产环境中。

模型评分服务

部署的模型根据实时客户请求（用于在线系统）或批处理（用于离线系统）进行预测。预测结果被记录到监控服务或在线评估存储中，以监控模型的预测性能，特别是概念/模型漂移。

您可以采用诸如金丝雀部署、影子模式部署和 A/B 测试等部署策略与评分服务一起使用。例如，您可以在生产环境中同时部署多个挑战者模型和冠军模型。它们都会收到来自客户端的相同预测请求，但只有冠军模型会返回预测结果。其他模型将将它们的预测结果记录到监控服务中。

性能监控和流水线反馈循环

流水线中的最后一个组件是监控组件，它对数据进行检查。它还跟踪收集的推断评估分数（模型指标或其他代理指标），以衡量生产环境中模型的性能。

一些监控组件还监控流水线的操作效率，包括：

• 流水线健康状况，
• API 调用，
• 请求超时，
• 资源使用情况等。

对于完全自动化的机器学习流水线，持续集成（CI）、持续交付（CD）和持续训练（CT）变得至关重要。流水线可以安排进行CI、CD或CT。它们还可以由以下方式触发：

• 1
  模型漂移，
• 2
  数据漂移，
• 3
  数据科学家按需触发。

如果您在生产中运行多个模型，自动化您的ML流水线将成为一个关键的生产力决策。

如何构建端到端的机器学习流水线

您可以按照以下顺序构建大多数流水线：

• 1
  将组件的代码实现定义为脚本中的模块化函数，或者重用现有的代码实现。
• 2
  将模块化脚本进行容器化，使其实现独立和分离。
• 3
  对实现进行打包，并将其部署在平台上。

模块化脚本

将组件定义为接受输入并返回输出的模块化函数是构建ML流水线的一种方式。它取决于您用于开发机器学习流水线的语言。使用特定领域语言（DSL），将这些组件链接在一起形成流水线。

以下是在Kubeflow Pipeline中定义ML流水线的DSL中编写的示例脚本：

import kfp.dsl as dsl

def my_pipeline_step(step_name, param1, param2, ...):
    return dsl.ContainerOp(
        name = step_name,
        image = '<path to my container image>',
        arguments = [
            '--param1', param1,
            '--param2', param2,
            ...
        ],
        file_outputs = {
            'output1' : '/output1.txt',
            'output2' : '/output2.json',
            ...
        }
    )

打包和容器
您可以决定使用Docker等容器工具或其他方法来确保您的代码可以在任何地方运行。
编排平台和工具
流水线编排平台和工具可以帮助将打包的脚本和容器管理为DAG或可以按顺序运行步骤的编排端到端工作流。
机器学习流水线工具
以下是一些机器学习流水线编排工具和平台的示例：



1

                Metaflow。

2

                Kedro流水线。

3

                ZenML。

4

                Flyte。

5

                Kubeflow流水线。            

Metaflow
Metaflow最初是Netflix的一个项目，是一个云原生框架，将ML堆栈的所有部分（从编排到版本控制、建模、部署和其他阶段）耦合在一起。Metaflow允许您将流水线指定为与工作流相关的计算的DAG。Netflix在Metaflow上运行数百到数千个机器学习项目，这就是它的可扩展性。
Metaflow与其他流水线框架的不同之处在于它可以将工件（如数据和模型）加载和存储为常规Python实例变量。只要具备Python的工作知识，任何人都可以使用它，无需学习其他特定领域语言（DSL）。

在文档中了解更多关于Metaflow的信息，并通过教程或资源页面开始使用。
Kedro
Kedro是一个用于构建模块化数据科学流水线的Python库。Kedro帮助您创建由可重用组件组成的数据科学工作流，每个组件都有“单一职责”，以加快数据管道流程、改善数据科学原型设计并促进管道的可重现性。

在本文中，您将学习如何使用Kedro构建ML流水线。
ZenML
ZenML是一个可扩展的开源MLOps框架，用于构建可移植的、生产就绪的MLOps流水线。它专为数据科学家和MLOps工程师设计，以便在开发生产过程中进行协作。

在文档中了解更多关于ZenML的核心概念。
Flyte
Flyte是一个用于大规模编排ML流水线的平台。您可以使用Flyte进行部署、维护、生命周期管理、版本控制和训练。您还可以与Feast、PyTorch、TensorFlow和whylogs等平台一起使用它来完成整个模型生命周期的任务。

这篇由Samhita Alla撰写的文章提供了Flyte在MLOps中应用的简化步骤。请查阅文档以开始使用。
Kubeflow Pipelines
Kubeflow Pipelines是一个用于在Kubernetes集群上直接构建和部署可移植、可扩展、可复现的端到端机器学习工作流的编排工具。您可以按照以下步骤定义Kubeflow Pipelines：
第1步：将每个组件的代码实现编写为可执行文件/脚本或重用预构建的组件。
第2步：使用特定于领域的语言（DSL）定义流水线。
第3步：构建和编译刚刚定义的工作流。
第4步：步骤3将创建一个静态的YAML文件，可以通过直观的Python SDK for pipelines来触发运行流水线。
Kubeflow的复杂性显而易见，而且开发迭代周期较慢，其他基于Kubernetes的平台如Flyte正在让构建流水线更加容易。但是，部署像Google Kubernetes Engine（GKE）这样的云托管服务可能更容易。
还有其他一些平台，例如Prefect和Argo，您也可以看一下。这篇文章可能很有用，因为它比较了10多个编排工具：最佳工作流和流水线编排工具。
DEMO：端到端ML流水线示例
在此示例中，您将根据Kaggle上臭名昭著的泰坦尼克号ML竞赛构建一个ML流水线。该项目使用机器学习创建一个模型，预测哪些乘客在泰坦尼克号沉船事故中幸存。
数据集还提供了有关泰坦尼克号上乘客的命运的信息，根据经济地位（类别）、性别、年龄和生存进行了总结。
先决条件

在此演示中，您将使用MiniKF在AWS上设置Kubeflow。Arrikto MiniKF是开始使用Kubeflow的最快最简便的方法。您还可以使用MiniKF在任何地方设置Kubeflow，包括您的本地计算机。您可以在文档中了解有关如何使用MiniKF在Google Cloud和本地计算机上设置Kubeflow的更多信息。
如果您还没有AWS帐户，请创建一个。
使用AWS Marketplace上的Arrikto MiniKF的费用为每小时0.509美元（根据撰写本文的时间）。此演示需要不到一小时完成，因此按照此演示不会花费超过3美元。
此演示使用的是Arrikto MiniKF v20210428.0.1，此版本安装了以下内容：

Kubeflow v1.3。
Kale v0.7.0。- 一种用于Kubeflow的编排和工作流工具，使您能够从笔记本开始运行完整的数据科学工作流。
Kubernetes (使用Minikube v1.22.0)。



在撰写本文时，演示步骤也适用于最新的Arrikto MiniKF v20221221.0.0。您可以按照官方文档中的教程，通过AWS Marketplace学习如何使用MiniKF部署Kubeflow。
如果您已经使用MiniKF部署了Kubeflow，让我们跳转到Kubeflow仪表板来设置项目：

要开始，请点击(1)“Notebooks”和(2)“+NEW SEVER”。
为您的笔记本服务器指定一个名称：

将其他选项保留为默认值（根据您的要求而定），然后在数据卷目录下点击“ADD VOLUME”：

现在您会看到一个新的数据卷添加，其名称与您为服务器指定的名称相同，并以“-vol-1/”作为后缀：

现在您可以启动笔记本服务器：

这可能需要几分钟的时间来设置，具体取决于您指定的资源数量。当您看到绿色的勾号时，点击“CONNECT”：

这将带您进入Jupyterlab启动器，在那里您可以创建一个新的笔记本并访问终端：

当您启动终端时，输入以下命令（记得输入您的数据卷名称）：
$ cd <输入您的数据卷名称>

$ git clone https://github.com/NonMundaneDev/layer-demo-kubeflow.git

(3) 启动`layer_kubeflow_titanic_demo.ipynb`笔记本：

在运行第一个代码单元格后，重新启动内核以使更改在当前内核中生效：

Kale可以将笔记本中的步骤编译成可以在Kubeflow Pipelines中运行的机器学习流水线。要将笔记本转换为ML流水线，
(1) 点击Kale图标，然后
(2) 点击启用：

Kale将自动检测应运行的步骤和应跳过的步骤，作为笔记本中的探索过程的一部分。在本笔记本中，Kale将所有步骤分类为一个组件，因为它们都需要输入并返回一个输出工件。
（1）您现在可以编辑流水线的描述和其他详细信息。完成后，
（2）点击“编译和运行”：

如果一切顺利，您应该会看到下面这样的可视化界面。点击“查看”旁边的“正在运行的流水线…”，会打开一个新的标签页：

您应该能够查看流水线的运行情况，并通过流水线界面查看刚刚通过Kale执行的Kubeflow Pipeline的有向无环图（DAG）：

现在，要查看模型在serving步骤中返回的结果，请点击“randomforest”步骤，进入“可视化”并向下滚动到“静态HTML”部分，查看最后一个单元格的预测结果：

在这种情况下，根据在notebook的serving步骤中传入的虚拟数据，模型预测该特定乘客在船难中没有生还。
您还可以通过以下步骤获取模型提供服务的URL端点：

在侧边栏中点击“模型”，观察到已经有一个模型正在提供服务。观察预测器、运行时和协议。点击模型名称。
您将看到一个仪表板，用于查看您正在生产中提供服务的模型的详细信息。
（1）使用指标和日志来监控您在生产环境中的模型并调试错误。您还可以查看
（2）“URL外部”和
（3）“URL内部”，这是您可以从任何其他服务请求或客户端访问模型的端点。“URL外部”可以重新路由到您的自定义URL。

目前，我们将通过“URL内部”端点通过终端访问模型。将端点复制并返回到您的Jupyterlab终端。使用以下命令将端点保存到变量中：
$ export MODEL_DEPLOYMENT_URL=<在此处输入您的内部URL端点>

$ curl --header "Content-Type: application/json; format=pandas-records"   --request POST   --data  '{"instances": [[3, 0, 4, 1, 2 ,3, 0, 1, 0, 8, 3, 6, 2]]}'  $MODEL_DEPLOYMENT_URL

您应该会得到与流水线notebook中相同的响应：

恭喜！您已成功构建了一个端到端的Kubeflow流水线。
与ML流水线相关的挑战

在使用机器学习流水线时，您可能会遇到以下一些挑战：



1. 基础设施和扩展需求。

2. 复杂的工作流程相互依赖。

3. 工作流程调度是一个困境。

4. 流水线的可重现性。

5. 实验跟踪。

基础设施和扩展需求
当您拥有卓越的基础设施时，机器学习流水线的优势才能得以实现。像Uber、Airbnb等公司都托管了他们的基础设施并且有预算来自行建设。对于依赖云基础设施推出产品的小型公司和初创公司来说，这是不切实际的。
如果您没有适当地监控云基础设施运行的数据、训练和生产流水线，可能会导致成本和费用呈指数级增长。您还可能遇到不同工作流程组件需要不同基础设施的情况。
机器学习流水线可以使您有效地运行实验并扩展，但是如果资源和预算有限，这个目的可能会受到限制。
复杂的工作流程相互依赖
由于工作流程步骤的复杂相互依赖关系，实现流水线工作流程可能会变得复杂，并且可能变得难以管理。
扩展复杂的工作流程相互依赖也可能会成为一个问题，因为某些组件可能需要比其他组件更多的计算资源。例如，模型训练可能需要比数据转换更多的计算资源。
工作流程调度困境
在机器学习流水线中进行工作流程调度并为错误和非计划情况提供弹性可能会非常棘手。当您使用工作流程调度程序时，很难指定作业失败时协调器应采取的所有操作。
流水线可重现性
在大规模运行数十个到数百个流水线时，这些流水线可能涉及各种数据转换、算法参数和软件依赖性，这可能会影响流水线的可重现性。

通常被忽视的是，用于生成模型的基础设施、代码和配置没有被正确地版本化，并且处于不可消耗、不可重现的状态。
— Ketan Umare，Union.ai的联合创始人兼首席执行官，在MLOps.community 2022的AMA会议上。

在其他情况下，您可能会使用特定的硬件配置构建流水线，运行在特定操作系统上，并具有不同的库依赖关系。但是，当将流水线编译运行在不同的环境中时，这些环境差异可能会影响机器学习流水线的可重现性。
构建机器学习流水线的最佳实践
通过研究社区讨论、与Brainly和Hypefactors等公司的工程师交流以及总结Netflix、Lyft、Spotify等公司的最佳实践，了解下面一些构建机器学习流水线的最佳实践。
跟踪您的机器学习流水线

我们在每次启动流水线时自动附加一个实验跟踪器，而我们的用户并没有注意到。对我们来说，这确保了至少跟踪一组最小参数…原则上，我们将实验跟踪视为应该与流水线一起使用的工具。我们建议使用流水线来跟踪您的实验，这样可以确保它们是可重现的。
— Simon Stiebellehner，在MLOps LIVE上的“Shipping Classic Software and Operating ML Models”中担任MLOps首席工程师和MLE Craft首席工程师。

您希望利用技术和方法使您的流水线具有可重现性和可调试性。这涉及到探索以下实践：

版本控制 – 用于管理依赖关系，包括代码、数据、配置、库依赖性、流水线元数据和工件，以便轻松跟踪和比较流水线版本。
实施系统治理 – 根据流水线中的步骤，您可以分析流水线运行的元数据和ML工件的血统来回答系统治理问题。例如，您可以使用元数据确定在特定时间哪个版本的模型正在生产中。
使用专用工具和框架 – 支持跟踪和管理流水线的工具和框架，如neptune.ai或MLflow，可以提供全面的跟踪和监控功能。

跟踪工具允许您：

记录实验结果，
可视化流水线组件，
记录步骤的详细信息以便团队成员之间进行协作，
监控执行过程中的流水线性能，更容易跟踪流水线随时间的演变，
管理流水线的进展。

这是关于ReSpo.Vision如何使用neptune.ai跟踪他们的管道的一个很好的案例研究。
ReSpo.Vision在体育数据分析中使用机器学习从单视角摄像机体育转播视频中提取3D数据。在这个过程中，他们运行了许多kedro管道。
ReSpo.Vision的CTO Wojtek Rosiński表示：“当我们在kedro中使用Neptune时，我们可以轻松跟踪在许多计算机上运行的管道的进度，因为我们经常同时运行许多管道，所以舒适地跟踪每个管道几乎是不可能的。有了Neptune，我们还可以轻松地使用不同的参数运行多个管道，然后通过用户界面进行结果比较。”
下面，您可以看到一个在Neptune用户界面中的示例。

Neptune原生集成了Kedro和ZenML等工具。但即使没有现成的集成，您也可以与任何其他的管道工具一起使用它。

了解如何开始使用neptune.ai
阅读完整的案例研究，了解ReSpo.Vision如何跟踪他们的实验和管道

将管道组件组合成较小的函数
使用流水线工具和SDK构建可重用组件（定义为较小的函数）来构建您的管道。以下是遵循ZenML管道工作流程的示例：
import numpy as np
from sklearn.base import ClassifierMixin
from sklearn.svm import SVC

from zenml.steps import step


@step
def svc_trainer(
    X_train: np.ndarray,
    y_train: np.ndarray,
) -> ClassifierMixin:
    """Train a sklearn SVC classifier."""
    model = SVC(gamma=0.001)
    model.fit(X_train, y_train)
    return model


这样，您可以通过构建自定义的或重用的预构建组件来实施您的工作流程。这样做可以使构建新的管道更加容易和快速，调试现有的管道，并将其与其他组织技术服务集成。

        不要在模块级别加载东西；这通常是一件坏事。您不希望模块加载花费很长时间并失败。
— Ketan Umare, Union.ai的联合创始人兼首席执行官，在MLOps.community 2022的AMA活动中。

下面是另一个使用Prefect编排工具定义为函数的步骤示例：
@task
def split_data(data: pd.DataFrame):
    # 将数据集随机分为70%用于训练和30%用于测试。
    X = data.drop("rented_bikes", axis=1)
    y = data.rented_bikes
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, train_size=0.7, test_size=0.3, random_state=42
    )
    return X_train, X_test, y_train, y_test

@task
def train_model(X_train: pd.DataFrame, y_train: pd.DataFrame):
    # 创建模型实例：GBRT（梯度提升回归树）
    model = GradientBoostingRegressor()
    # 模型训练
    model.fit(X_train, y_train)
    return model


编写管道测试
另一个最佳实践是确保您构建了一个测试套件，涵盖了管道的每个方面，从组成组件的函数到整个管道的运行。如果可能的话（根据用例），愿意自动化这些测试。

        为了确保模型在底层训练或服务容器图像持续变化时继续按预期工作，我们为LyftLearn Serving提供了一系列独特的测试，称为模型自测试。
— Mihir Mathur, Lyft的产品经理，在“使用LyftLearn Serving进行数百万实时决策”的博客中，2023年。

将管道组件组合成较小的函数可以更容易进行测试。以下是Lyft的模型自测试示例，其中他们在名为`test_data`的函数中指定了模型输入和期望输出的少量样本：
class SampleNeuralNetworkModel(TrainableModel):

    @property

    def test_data(self) -> pd.DataFrame:

        return pd.DataFrame(

            [

                # 输入 `[1, 0, 0]` 应该生成接近 `[1]` 的输出

                [[1, 0, 0], 1],

                [[1, 1, 0], 1],

            ],

            columns=["input", "score"],

        )

在大多数情况下，当您的堆栈和设置使本地测试变得不可能时，您的用户很可能最终会在生产环境中进行测试。将您的步骤容器化可以使您在部署到生产环境之前更容易地在本地或其他环境中测试您的流水线。
您应该编写哪些流水线测试？在他的文章中，Eugene Yan列出了有效的流水线测试应该是什么样子的范围图，包括单元测试、集成测试、功能测试、端到端测试等等。请查看详细的文章。
结论
构建端到端的机器学习流水线是现代机器学习工程师的关键技能。通过遵循最佳实践，如彻底的测试和验证、监控和跟踪、自动化和调度，您可以确保流水线的可靠性和效率。
通过对每个流水线阶段的组件、结构和挑战有坚实的理解，您可以构建健壮且可扩展的流水线，以简化您的机器学习工作流程。
祝您流水线工作顺利！

参考资料

《Kubeflow流水线入门》- YouTube
《使用Kedro构建和管理数据科学流水线》- neptune.ai
《Metaflow如何成为Netflix喜爱的数据科学框架》- Julie Amundson
https://mlops-community.slack.com/archives/C02NLLRUVN3/p1660842600835749
《Flyte：简化的MLOps》- MLOps Community
https://mlops-community.slack.com/archives/C02NLLRUVN3/p1660844981495249