构建一个简单的端到端MLOps工作流与 Google Vertex AI

在快速发展的机器学习领域，部署模型到生产环境并高效维护这些模型至关重要。这让我开始探索 MLOps — 结合了机器学习（ML）与 DevOps 的一套实践方法，旨在自动和简化整个机器学习的生命周期。2021 年，在我早期的职业生涯中构建了一个类似但稍微更复杂的管道，这让我非常期待再次审视这一主题，以便跟上最新的发展。

在这篇博客文章中，我将一步步带您了解如何使用Google Cloud的Vertex AI构建一个端到端的MLOps流水线。这个流水线自动化了数据获取、模型训练、评估和模型部署。

不论你是希望将模型投入实际应用的机器学习从业者，还是对MLOps（机器学习运维）感兴趣的人，本文希望提供一个使用Vertex AI构建和部署这些模型的实用说明。

你可以点击这里: https://github.com/whiteking64/mlops-gcp/tree/v1.0 查看完整实现。

MLOps（机器学习运维）是一套旨在可靠且高效地在生产环境中部署和维护机器学习模型的方法。它更专注于解决机器学习系统特有的挑战，例如数据版本管理、模型训练、验证、部署及监控。

MLOps的关键方面有：

• 自动化——在机器学习生命周期中自动化重复性任务。
• 持续集成和部署（CI/CD）——确保代码或数据的更改能实现可靠的、可重复的部署流程。
• 监控与反馈——持续地监控模型性能，并根据反馈来改进模型。

Google Vertex AI 是一个集成 Google Cloud 服务的统一平台，用于构建 ML 模型。它通过提供覆盖整个 ML 生命周期的工具和服务，简化了开发和部署 ML 模型的过程。

Vertex AI的关键特性

• 工作台：使用集成的 Jupyter 笔记本来探索数据并开发模型。
• 流水线：主要使用 Kubeflow 流水线来编排 ML 工作流程。
• 训练：大规模训练模型并支持自定义训练和超参数调整。
• 预测：部署模型以提供在线或批量预测。
• 模型注册表：在集中式存储库中管理和版本化模型。
• 特征存储：管理、提供和共享 ML 特征资源。

好处 ：

• 一体化平台：通过整合所有必要的工具来简化机器学习的工作流程。
• 可扩展性：根据任务需求轻松扩展所需资源。
• 管理基础设施：利用 Google 的管理服务来减少运营成本。

该项目的目标是构建一个完整的MLOps流水线，具体包括从数据预处理到模型部署的各个环节：

• 从数据源（本例中的 BigQuery）抓取数据。
• 使用 Hugging Face Transformers 训练和评估测试文本分类模型。
• 将模型部署到 Vertex AI 以提供预测功能。
• 使用 Kubeflow Pipelines (KFP) 自动化整个过程。

注意：此流程展示了一个简化的架构。在实际生产环境中，需要更多额外的组件和更强大的设置。

场景：

• 新闻分类：比如说使用AG新闻数据集将新闻文章分类到预设的类别。

使用的技术和工具：

• Vertex AI ：用于模型训练、部署和提供。
• Kubeflow Pipelines ：用于编排机器学习流程。
• Docker 和 Docker Compose ：用于应用容器化部署。
• BigQuery ：用于数据存储和查询。
• Hugging Face Transformers ：用于模型开发。

这个管道系统包括几个关键组件。

1. 数据准备阶段：加载和预处理数据集，然后将其上传到BigQuery。
2. 工作流编排阶段：定义数据获取、模型训练和部署的组件，并使用Kubeflow Pipelines编排工作流。
3. 模型训练：训练一个用于文本分类的DistilBERT模型，并通过评估来检验模型的性能。
4. 模型部署：将模型部署到Vertex AI平台，并设置预测端点。

让我们来看看管道中的每个环节，看看它们是怎么做的。

目标：加载数据，对数据集进行预处理，并将其上传到BigQuery以便在其中存储和查询。

步骤如下：

1. 加载数据集 :
   - 使用了来自Hugging Face的datasets库中的AG News数据集。
   - 选取了2,000条记录作为演示示例。
2. 数据预处理 :
   - 将数值标签映射为对应的类别名称（例如，0映射为World）。
   - 添加了一个timestamp字段以模拟实际最近的数据条目。
3. 上传到BigQuery :
   - 使用服务账号凭证配置了BigQuery客户端的设置。
   - 将预处理后的DataFrame上传到了一个BigQuery表内。
   注意：虽然上传到BigQuery并非必需，但在许多实际应用中非常普遍。在这个项目中，我们使用BigQuery来模拟典型的存储和查询系统。或者，你也可以将数据存储到Google Cloud Storage（GCS），这也是一种在ML管道中常见的数据存储方式。

下载和上传数据 (准备数据集的脚本)：

    dataset = load_dataset("ag_news")  
    subset = dataset["train"].select(range(size))  
    df = pd.DataFrame(subset)  

    label_mapping = {0: "世界", 1: "体育", 2: "商业", 3: "科学与技术"}  
    df["label"] = df["label"].map(label_mapping)  

    # 添加时间戳以模拟近期数据  
    df["timestamp"] = pd.date_range(end=datetime.datetime.now(), periods=size)  

    # 将数据上传到Google的BigQuery数据仓库  
    df.to_gbq(  
        destination_table=f"{config['dataset_id']}.{config['table_id']}",  
        project_id=project_id,  
        if_exists="replace",  # 在表已存在的情况下替换它  
    )

timestamp字段在这个流程中非常重要，特别是在多次运行的过程中考虑时间因素时。原因如下：

• 模拟真实世界的情景：这样一来，通过添加时间戳，我们能够模拟一个真实世界的情景，在这种情景中，数据持续地添加到我们的数据集中。这使我们能够测试和验证我们处理新数据的能力。
• 逐步学习：这样一来，在实际应用中，我们通常希望使用最新数据重新训练模型。添加时间戳使我们能够轻松查询和获取新数据，从而实现逐步学习。
• 性能监控：这样一来，通过在不同时期的数据显示，我们能够监控模型性能的变化随时间，帮助我们检测概念漂移或数据分布的变化。

在后续的管道运行过程中，您可以修改数据获取组件，让它只抓取某个时间戳之后的数据，确保每次训练都能获取最新的数据，不会重复。

执行之后，你就可以看到数据已经被正确地存储了，在BigQuery中。

目标：利用Kubeflow Pipelines来定义和调度ML的工作流程。

定义管道流程（main.py）：

    @dsl.pipeline(  
        name="文本分类管道",  
        description="一个训练和部署文本分类模型的管道",  
    )  
    def text_classification_pipeline():  
        fetch_data_task = fetch_data_from_bigquery(...)  
        train_task = train_model(input_dataset=fetch_data_task.outputs["output_dataset"])  
        deploy_task = deploy_model(model=train_task.outputs["output_model"])

编译并运行流程:

compiler.Compiler().compile(  
    pipeline_func=text_classification_pipeline,  
    package_path="text_classification_pipeline.json",  
)  

job = pipeline_jobs.PipelineJob(  
    display_name="文本分类流水线",  
    template_path="text_classification_pipeline.json",  
    pipeline_root=f"gs://{config['bucket_name']}/pipeline_root",  # 设置管道根目录，用于保存运行时生成的文件
)  
job.run(service_account=config["service_account"])  # 使用指定的服务账户运行流水线作业

在 Vertex AI 的“工作流”页面上，您可以看到所有执行过的管道。

目标：在拿到的数据集上训练一个DistilBERT模型来并对其性能进行评估。

关键步骤如下：

1. 数据加载 :
   - 加载了由前一个组件生成的数据集文件。
   - 对类别标签应用了标签编码。
2. 模型准备 :
   - 初始化了 DistilBERT 分词器和模型。
   - 配置了训练参数设置。
3. 训练和评估 :
   - 使用 Hugging Face 的 Trainer API 进行训练。
   - 在验证集上评估了模型。
   - 计算了诸如准确率、精确度、召回率和 F1 值等指标。
4. 保存模型 :
   - 保存了训练好的模型、分词器和标签编码器，以便后续部署。

代码片段（训练代码）：

    tokenizer = AutoTokenizer.from_pretrained("distilbert-base-uncased")  
    model = AutoModelForSequenceClassification.from_pretrained(  
        "distilbert-base-uncased", num_labels=num_labels  
    )  

    # 准备训练和验证数据集  
    train_encodings = tokenizer(train_texts, truncation=True, padding=True)  
    val_encodings = tokenizer(val_texts, truncation=True, padding=True)  

    # 初始化 Trainer  
    training_args = TrainingArguments(...)  
    trainer = Trainer(  
        model=model,  
        args=training_args,  
        train_dataset=train_dataset,  
        eval_dataset=val_dataset,  
    )  
    trainer.train()  

    # 评估模型并保存结果  
    predictions = trainer.predict(val_dataset)  
    ...  
    model.save_pretrained(output_model.path)  
    tokenizer.save_pretrained(output_model.path)

我们的目标是将这个训练好的模型部署到Vertex AI来提供预测结果。

部署步骤：

1. 模型上传：
   - 将模型工件上传到Vertex AI模型注册表。
   - 指定了自定义的Docker镜像用于部署服务。
2. 端点创建：
   - 检查具有相同名称的端点是否存在并删除了以避免冲突。
   - 创建新的端点以供模型使用。
3. 模型部署：
   - 将模型部署到指定的机器类型和缩放选项的端点。

代码段（部署流程）：

    # 上传模型
    uploaded_model = aiplatform.Model.upload(  
        display_name=model_display_name,  
        serving_container_image_uri=serving_image_uri,  
        artifact_uri=model.uri,  
        serving_container_ports=[8080],  
    )  

    # 创建端点
    endpoint = aiplatform.Endpoint.create(display_name=endpoint_display_name)  
    # 部署模型到端点
    deployed_model = uploaded_model.deploy(  
        endpoint=endpoint,  
        machine_type="n1-standard-4",  
        min_replica_count=1,  
        max_replica_count=1,  
    )

模型部署需要一些时间，完成之后，您就可以在模型注册页面看到模型已经准备就绪。

点击名字，你会发现与模型关联的端点。

在左侧窗格中导航到“在线预测”，你可以看到所有成功或未成功的端点。显然，失败的端点不能被删除。

这是整个工作流！你可以看到DAG，并且所有节点都成功（绿色的勾选）。小图标是组件的输出。在右边，有一个工作流摘要，包括持续时间和指标。就我而言，管道完成耗时1小时。

我们来看一个节点。选择“deploy-model”节点，你就会在右边看到详细信息。按照我们的实现，输入参数已经正确设置。

看看训练节点，你可以看到输出结果。看来是进展顺利！

如果某个节点出了问题，你可以点击页面底部的“日志”来查看详情。

输出文件会被保存到云存储的管道根目录中。

目的：发送一些预测请求样本来测试已部署的模型。

步骤如下：

1. 认证 :
   - 确保 gcloud CLI 已通过适当权限的身份进行了认证。
2. 发送请求 :
   - 准备了一个包含示例文本的 JSON 数据。
   - 向模型的预测端点发送了一个 POST 请求。
3. 解读响应 :
   - 收到了预测结果，包括预测类别及其概率。

示例请求示例 (sample-request.sh）：

    curl -X POST \  
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \  
    -H "Content-Type: application/json" \  
    "https://us-central1-aiplatform.googleapis.com/v1/projects/$project_id/locations/$location/endpoints/$endpoint_name:predict" \  
    -d '{  
        "instances": [  
            {"text": "示例新闻文章文本……"},  
            ...  
        ]  
    }'

这应该看起来像下面的链接（我的 GitHub 项目主页，再次说明）：

在构建管道时，我遇到了一些挑战。最棘手的一个是模型的部署。

复杂性来自理解多个概念，比如模型、端点、它们的关系以及配置。

我还需要准备一个自定义的容器。有一些预构建的容器，其中包含了PyTorch，其中模型通常是以.mar格式保存的。然而，我使用的是Hugging Face的模型并用PyTorch保存为.safetensors格式的模型。将其转换为其他格式证明很棘手，而且我不确定这会对其它伴随文件如编码器造成什么影响。因此，我选择了自定义容器来处理这个问题。Vertex AI的新功能“模型花园”（Model Garden）提供了一种替代方案，您可以在其中指定一个Hugging Face URL来“从Hugging Face上部署”。在这个项目中，这个管道没有将我们微调的模型推送到HF，因此这不是一个选项，但未来值得探索一下。

值得一提的是，你可以通过使用Python Vertex AI SDK中的LocalModel上传模型。我尝试过这样做，但截至2024年9月似乎存在一个小问题，所以我决定暂时不这样做。

https://github.com/GoogleCloudPlatform/vertex-ai-samples/issues/2833 (这是一个关于Vertex AI样本的问题页面)

另一个小小的挑战是训练部分。如果你看过我的实现，你会注意到我没有使用GPU来进行模型训练。虽然你可以调整机器配置，比如选择不同的加速器等，但遇到了配额限制。作为一个小小的演示，仅用CPU训练模型就已经足够，这实际上不会花太多时间。

基于之前讨论的挑战，我们的 MLOps 管道在改进方面还存在一些局限性。

• 架构和可扩展性：当前的单体结构限制了模块化和可扩展性。未来版本应将管道分解为微服务，并利用分布式计算以提高效率。
• 监控和错误处理：缺乏全面的监控、警报和错误处理机制限制了管道的可靠性。集成强大的日志记录和监控工具，并实现重试机制将显著提高操作稳定性。
• 模型开发和实验：训练部分缺乏模块化和超参数调整能力。未来的改进应集中在分离预处理、训练和评估模块，并结合Vertex AI的超参数调整任务来优化模型性能。
• 部署策略：当前的部署会覆盖现有模型，导致停机，并使生产环境中的回退复杂化。未来的改进应包括实现模型版本控制以管理同时存在的多个模型，并采用更安全的部署方式，例如金丝雀发布。

欢迎在LinkedIn或GitHub上关注我的项目和见解，以了解更多项目和见解.