使用 Prefect 编排机器学习工作流

在本教程中，我们将通过介绍 Prefect 的关键特性、优势和组成部分来深入了解该工具。我们将演示如何创建任务与流程（flows）、管理部署，并执行流程运行。最后，我们会指导您将工作流部署到 Prefect Cloud，并使用云端基础设施运行它。

如果您对如何使用 GitHub Actions 和数据版本控制自动化机器学习流水线感兴趣，建议学习我们的机器学习的 CI/CD课程。

什么是 Prefect？

Prefect 是一款功能强大的开源工作流编排工具，可让用户用 Python 代码设计、监控并响应数据与机器学习流水线。

Prefect 内置了自动重试、调度、缓存等多种特性，是一款能够帮助您构建稳健且动态工作流的强大工具。

图片来源

借助 Prefect，您可以轻松将任意 Python 函数转换为可观测、可编排的工作单元，只需在代码中添加几个装饰器即可。

Prefect 非常适合在组织内部标准化工作流的开发与部署。它为工作流提供完整的可观测性，并允许用户在无需严格的DAG或样板代码的情况下管理其代码。

Prefect 关键特性

轻量化

只需一条命令，用户即可轻松搭建自己的编排服务器用于开发。这一精简流程简化了本地开发，并确保顺利部署到生产环境。

直观的 UI 与仪表盘

Prefect 提供总体直观的用户界面，为运行、监控、调试和管理工作流提供了美观且易用的方式，从而提升用户体验。

功能全面

Prefect 提供调度、重试、原生机密管理、扩展能力、并发控制、灵活的基础设施与存储配置选项，以及健壮的日志功能等。

开源的灵活性

用户可以选择自托管开源版的 Prefect 服务器，或使用 Prefect Cloud 以获得高级特性、高可用性和更强的安全性。

Prefect Cloud

Prefect Cloud 在开源服务器的全部能力之上，面向托管环境提供了自动化、事件订阅、事件、webhook、工作区、组织等更多特性。您可以免费使用 Prefect Cloud，也可以付费解锁更强大的功能。

纯 Python

Prefect 允许用户通过一个装饰器就开始进行工作流编排，过程精简，无需不必要的抽象，即可高效管理工作流。

通知

将有关 Prefect 工作流和流水线的信息发送到多个消息服务，如 Slack、Discord、电子邮件、Microsoft Teams 等。当工作流失败时也可触发告警。

Prefect 的核心组件

本节将介绍用于构建任务与流程、部署与执行，以及管理它们的关键组件。在进入 Python 教程前，了解这些组件至关重要。

Prefect Task

Prefect 任务（Task）是使用 @task 装饰器修饰的 Python 函数，表示Prefect 工作流中的离散工作单元。我们还可以通过可选参数（如名称、描述、标签、缓存设置、重试次数等）自定义任务装饰器。

在下面的代码中，我们用 @task 装饰器创建了三个机器学习任务，它们可以在多个流程中复用。

@task
def train_model(X_train, X_test, y_train):
    # Selecting the best features
    ...

    # Train the model
    ...

    return model


@task
def get_prediction(X_test, model: LogisticRegression):
    ...
	return prediction

@task
def evaluate_model(y_test, prediction: pd.DataFrame):
    ...技术支持

Prefect Flow

Prefect 流程（Flow）是使用 @flow 装饰器修饰的 Python 函数，用于封装工作流逻辑。这使用户能够更加轻松灵活地定义、配置并执行复杂的数据流水线。

在下面的代码中，我们使用 @flow 装饰器创建了一个流程。该流程会按顺序运行所有任务，并将前一任务的输出作为后一任务的输入。

@flow
def ml_workflow():
    model = train_model(X_train, X_test, y_train)
    predictions = get_prediction(X_test, model)
    evaluate_model(y_test, predictions)

if __name__ == "__main__":
    ml_workflow()技术支持

Prefect 部署

部署（Deployments）是存储在本地服务器或云端的流程，包含用于远程编排工作流的重要信息，例如调度与执行细节。

Prefect 工作池

工作池（Work Pools）在编排与执行环境之间充当中介，支持高效调度与执行流程运行。用户可以设置本地运行工作流的 Agent（工作器），或利用云端基础设施实现无缝的工作流执行。

类似于专用计算实例，工作池能高效管理任务分配并对任务进行优先级排序，提供灵活的执行环境选择，以实现最佳的工作流性能与可扩展性。

编排机器学习工作流

本节我们将使用 Prefect 创建、测试、部署并执行一个简单的机器学习工作流。我们还会设置通知，在 Discord 服务器上接收有关流程运行状态的提醒。

安装与设置 Prefect

使用 PIP 安装该 Python 包

pip install -U prefect技术支持

如果您想使用 Prefect 的最新版本，可以直接从 GitHub 获取代码。

pip install -U git+https://github.com/PrefectHQ/prefect技术支持

创建一个 Prefect Flow

我们将预处理来自 Kaggle 的 Bank Churn 数据集，然后使用 scikit-learn 训练并评估随机森林分类器模型。

我们将使用的代码是对使用 Scikit-learn Pipelines 精简您的机器学习工作流一文中代码的修改版。如果您想进一步了解模型的训练方式，可查看该教程。我们的机器学习 Prefect Flow 包含七个任务：

load_data：使用 pandas 加载并处理 CSV 文件
preprocessing：为训练准备数据，填补缺失值、编码与缩放
data_split：将数据划分为测试集与训练集
train_model：选择最优特征并训练模型
get_prediction：使用测试集生成预测
evaluate_model：计算准确率与 F1 分数
save_model：使用 skops 保存模型权重

我们现已将所有必要任务组装进名为 ml_workflow 的 @flow 函数中。同时在装饰器中加入了额外参数以启用详细日志。

ml_workflow 函数以数据文件位置为输入，然后将各个任务串联起来。等我们在 Prefect 服务器上部署并执行该流程后，就能更直观地查看此工作流。

您还可以使用 MLflow 跟踪机器学习实验。请学习MLflow 入门课程，了解 MLflow 跟踪、项目、模型及模型注册表。

import pandas as pd
import skops.io as sio
from prefect import flow, task
from sklearn.feature_selection import SelectKBest, chi2
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, f1_score
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler, OrdinalEncoder


@task
def load_data(filename: str):
    bank_df = pd.read_csv(filename, index_col="id", nrows=1000)
    bank_df = bank_df.drop(["CustomerId", "Surname"], axis=1)
    bank_df = bank_df.sample(frac=1)
    return bank_df


@task
def preprocessing(bank_df: pd.DataFrame):
    cat_col = [1, 2]
    num_col = [0, 3, 4, 5, 6, 7, 8, 9]

    # Filling missing categorical values
    cat_impute = SimpleImputer(strategy="most_frequent")
    bank_df.iloc[:, cat_col] = cat_impute.fit_transform(bank_df.iloc[:, cat_col])

    # Filling missing numerical values
    num_impute = SimpleImputer(strategy="median")
    bank_df.iloc[:, num_col] = num_impute.fit_transform(bank_df.iloc[:, num_col])

    # Encode categorical features as an integer array.
    cat_encode = OrdinalEncoder()
    bank_df.iloc[:, cat_col] = cat_encode.fit_transform(bank_df.iloc[:, cat_col])

    # Scaling numerical values.
    scaler = MinMaxScaler()
    bank_df.iloc[:, num_col] = scaler.fit_transform(bank_df.iloc[:, num_col])
    return bank_df


@task
def data_split(bank_df: pd.DataFrame):
    # Splitting data into training and testing sets
    X = bank_df.drop(["Exited"], axis=1)
    y = bank_df.Exited

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.3, random_state=125
    )
    return X_train, X_test, y_train, y_test
    # Identify numerical and categorical columns


@task
def train_model(X_train, X_test, y_train):
    # Selecting the best features
    KBest = SelectKBest(chi2, k="all")
    X_train = KBest.fit_transform(X_train, y_train)
    X_test = KBest.transform(X_test)

    # Train the model
    model = LogisticRegression(max_iter=1000, random_state=125)
    model.fit(X_train, y_train)

    return model


@task
def get_prediction(X_test, model: LogisticRegression):
    return model.predict(X_test)


@task
def evaluate_model(y_test, prediction: pd.DataFrame):
    accuracy = accuracy_score(y_test, prediction)
    f1 = f1_score(y_test, prediction, average="macro")

    print("Accuracy:", str(round(accuracy, 2) * 100) + "%", "F1:", round(f1, 2))


@task
def save_model(model: LogisticRegression):
    sio.dump(model, "bank_model.skops")


@flow(log_prints=True)
def ml_workflow(filename: str = "train.csv"):
    data = load_data(filename)
    prep_data = preprocessing(data)
    X_train, X_test, y_train, y_test = data_split(prep_data)
    model = train_model(X_train, X_test, y_train)
    predictions = get_prediction(X_test, model)
    evaluate_model(y_test, predictions)
    save_model(model)


if __name__ == "__main__":
    ml_workflow()技术支持

要测试我们的 Prefect Flow，请在终端中运行该 Python 文件。

$ python main.py技术支持

流程已成功完成。它已输出模型指标并保存了模型权重。

如果您是机器学习新手，想学习如何预处理数据、训练模型、优化、评估并保存模型，可以考虑学习Python 机器学习科学家职业路径课程。该课程将从零开始教您所需的一切。

部署流程

手动构建并运行一个机器学习流水线并不复杂，即便使用 scikit-learn 的 pipelines 也能做到。

不过，使用 Prefect 有诸多好处，包括：

对流水线进行监控，
使用 CRON 进行调度，
失败时自动重试，
启用日志与可观测性，
接收通知，
并创建可在无人干预下运行的自动化工作流。

现在，我们将把工作流部署到本地 Prefect 服务器，实现自动化。

首先，我们需要构建“部署（Deployment）”，指定文件与流程函数名称。我们还为部署添加了“dev”标签。

注意：在 Prefect 中，标签可用于在生态系统内对流程、任务和部署进行分类与组织。

$ prefect deployment build main.py:ml_workflow -n 'ml_workflow_bank_churn' -a --tag dev技术支持

部署已成功创建。接下来它会给出使用 Prefect Agent 执行该流程的说明。

Found flow 'ml-workflow'
Deployment YAML created at 
'C:\Repository\GitHub\ML-Workflow-Orchestration-With-Prefect\ml_workflow-deployment.yaml'.
Deployment storage None does not have upload capabilities; no files uploaded.  Pass --skip-upload to 
suppress this warning.
Deployment 'ml-workflow/ml_workflow_bank_churn' successfully created with id 
'ce30d1c7-f454-4064-870c-429a8039c194'.

To execute flow runs from this deployment, start an agent that pulls work from the
'default-agent-pool' work pool:
$ prefect agent start -p 'default-agent-pool'技术支持

在本地运行部署

Prefect Agent 负责监控工作池，识别可用的流程运行，并将其部署到执行环境。

单个工作池可拥有多个 Agent，每个 Agent 都有唯一 ID，以避免重叠并确保顺畅扩展。

我们现在在新终端中以默认工作池名称启动 Prefect Agent。

$ prefect agent start -p 'default-agent-pool'技术支持

转到另一个新的终端，使用“flow_function/部署名称”来运行部署。您也可以仅使用部署名称来执行流程。

$ prefect deployment run 'ml-workflow/ml_workflow_bank_churn'技术支持

它已为该部署创建了一个流程运行，会检查可用的工作器并开始运行工作流。

Creating flow run for deployment 'ml-workflow/ml_workflow_bank_churn'...
Created flow run 'cryptic-potoo'.
└── UUID: 77833ee3-0f7e-42f9-8002-0a562c7d3ed4
└── Parameters: {}
└── Scheduled start time: 2024-03-14 21:30:40 PKT (now)
└── URL: <no dashboard available>技术支持

切换到我们启动 Agent 的终端以查看流程运行的日志。Prefect Agent 已成功执行流程中的所有任务。

这是一个简单示例。我们也可以在构建部署时提供更多参数来对工作流进行调度或自动化。

Prefect UI

到目前为止，我们一直在终端中观察工作流。现在，让我们启动 Prefect 仪表盘，在那里可以查看所有流程运行、部署、工作池和通知。

在新终端中启动 Prefect 服务器，然后点击提供的本地链接进入仪表盘。

$ prefect server start技术支持

要查看此前的所有流程运行，请点击左侧面板中的 "Flow" 选项。您可能会注意到，部分流程由于缺少工作池而未执行，另一些则是在试验阶段未使用 "dev" 标签执行的。

点击 Flows 选项会显示所有可用流程。单个部署中可以存在多个流程。

在 Deployment 部分，我们可以查看当前部署、其活动与标签。在此处可通过快速运行选项直接运行流程，或使用界面进行调度。

在 Work Pools 中，您会看到我们的默认 Prefect Agent。您也可以使用 Prefect CLI 设置云端工作池。

添加 Discord 通知

让我们进入有趣的部分。

没有人希望全天候盯着工作流。我们更希望在喜欢的消息应用中接收工作流状态通知，这样就能快速处理问题。

为实现这一点，我们将创建一个 Discord webhook。借助 webhook 的 ID 和令牌，我们将设置 Prefect 通知，把所有日志信息发送到 Discord 服务器。

要创建 webhook，请前往您的 Discord 服务器设置，点击“Integrations”，选择“View Webhooks”，然后点击“New Webhook”。

更改显示图片、名称与频道，然后复制 Webhook URL。

Webhook URL 由两部分组成，即 Webhook Id 与 Webhook token：https://discord.com/api/webhooks/<webhook_id>/<webhook_token>。要创建 Prefect 通知，我们需要从该链接中手动分离出这两部分。

要在 Prefect UI 中创建 Discord 通知，进入“Notifications”选项，点击顶部的“+”按钮。然后选择所需的运行状态、标签、Discord webhook 与通知类型。

接着，填写 Webhook ID 和 Webhook token，并点击蓝色的“Create”按钮。

我们需要执行部署，才能在 Discord 服务器的主频道中看到通知。

$ prefect deployment run 'ml-workflow/ml_workflow_bank_churn'技术支持

太棒了。用同样的方法，我们还可以设置 Microsoft Teams、Slack、短信和电子邮件等多种消息服务，甚至可以创建自定义 webhook。

学习用 Python 设计机器学习工作流课程，掌握能够经受时间考验的流水线构建方法。

在云端运行部署

在本地部署了机器学习流程之后，接下来要将其部署到 Prefect Cloud。相比本地部署，这一过程更为轻松，因为您无需启动服务器或 Agent，也无需通过 CLI 工具执行运行。一切都由 Prefect 托管管理。

将代码推送到 GitHub 仓库

在部署当前流程之前，需要注意：它会尝试访问云端不可用的本地目录中的文件和数据，从而报错。为了让 Prefect Cloud 能够访问这些文件，我们必须将代码创建并推送到一个公共 GitHub 仓库。

这可以通过 GitHub 网站轻松完成，或使用 VSCode 的 Git 扩展在界面中创建远程仓库并推送代码。

以下是我们用于云端部署的项目仓库：kingabzpro/ML-Workflow-Orchestration-With-Prefect

设置 Prefect Cloud

前往 https://app.prefect.cloud/auth/login 创建 Prefect Cloud 账户，然后使用 Prefect CLI 工具登录。

$ prefect cloud login技术支持

选择“Log in with a web browser”，并按照步骤将本地客户端与 Prefect Cloud 进行认证。

我们将创建一个由 Prefect 托管的工作池，以充分利用 Prefect Cloud 的基础设施，而不是使用本地运行的 Agent。

$ prefect work-pool create DC-work-pool --type prefect:managed技术支持

对于免费账户，Prefect 每月提供 10 小时的免费计算配额，供您试验与练习。

部署 Python 代码

现在我们将以编程方式把流程部署到 Prefect Cloud。

创建一个名为 deployment.py 的新文件。

为 Prefect 流程函数设置参数：

source：包含流程文件的 GitHub 仓库链接。
entrypoint：文件名与流程函数名。
work_pool_name：提供由 Prefect 托管的工作池名称。
tags：用于分类。
job_variables：用于安装运行代码所需的 Python 包。如果您使用的是类似本地 Prefect Agent 的混合型工作池，则无需提供 job_variables。

from prefect import flow

if __name__ == "__main__":
    flow.from_source(
        source="https://github.com/kingabzpro/ML-Workflow-Orchestration-With-Prefect.git",
        entrypoint="main.py:ml_workflow",
    ).deploy(
        name="first-prefect-deployment",
        work_pool_name="DC-work-pool",
        tags=["dev"],
        job_variables={"pip_packages": ["pandas", "skops", "scikit-learn"]},
    )技术支持

运行该 Python 代码。

$ python deployment.py技术支持

流程已部署至 Prefect Cloud。可以通过 CLI 或 Prefect Cloud 仪表盘执行。

在云端运行流程

要运行该流程，请访问网站：https://app.prefect.cloud/。然后进入“Deployments”部分，点击部署右侧的竖向三点按钮，选择“Quick run”。

要查看运行状况，请进入“Flow Run”部分，点击最新一次运行以查看工作流图。

流程图概览了各任务之间的连接关系以及每个过程耗时。

若需查看详细日志，请向下滚动，查看带时间戳的所有任务执行记录。

在“Work Pools”部分，我们可以清楚地看到，在免费用户 10 小时的限额中，我们已使用了 0.1 小时。

查看机器学习工作流指南，其中包含信息图，帮助您理解项目搭建、数据准备、建模与部署各环节所涉及的任务。

总结

搭建一个单一的机器学习工作流并不难。但在真实场景中，您可能需要同时运行多个机器学习与数据工作流。Prefect 的优势正在于此：无论是选择在本地还是云端环境运行 Prefect 服务器或工作池，它都提供了良好的可扩展性与灵活性。它是进行调度、自动化与管理工作流的出色工具。

在本教程中，我们介绍了 Prefect 生态系统的主要功能与特性，并学习了如何在本地与云端部署机器学习流程。若要继续您的 MLOps 之旅，下一步请通过“机器学习 CI/CD 入门指南”学习如何为机器学习工作流构建 CI/CD。

Prefect 只是 MLOps 领域中用于工作流编排的工具之一。若要了解用于数据版本管理、特征库、模型测试、模型部署与服务、模型监控、运行时引擎以及端到端 MLOps 的其他重要工具，请阅读文章2024 年必知的 25 款 MLOps 工具。您也可以阅读我们的Prefect 与 Airflow 对比，看看两者如何各展所长。

作者

Abid Ali Awan

LinkedIn Twitter

作为一名持证的数据科学家，我热衷于利用前沿技术打造创新的机器学习应用。凭借在语音识别、数据分析与报告、MLOps、对话式人工智能以及自然语言处理方面的扎实背景，我不断打磨构建智能系统的能力，力求带来切实影响。除技术专长外，我也擅长沟通，能够将复杂概念提炼为清晰、简明的表述。因此，我成为数据科学领域备受关注的博主，与不断壮大的数据专业人士社区分享洞见与实践经验。目前，我专注于内容创作与编辑，借助大语言模型打造有力且吸引人的内容，帮助企业与个人更好地发挥数据价值。

主题

机器学习 Python