如何将Apache Hudi应用于机器学习

1. 引入

如果要将AI嵌入到企业计算系统中，企业必须重新调整其机器学习（ML）开发流程以使得数据工程师、数据科学家和ML工程师可以在管道中自动化开发，集成，测试和部署。本博客介绍了与机器学习平台进行持续集成（CI），持续交付（CD）和持续培训（CT）的平台和方法，并详细介绍了如何通过特征存储（Feature Store）执行CI / CD机器学习操作（MLOps）。以及特征存储如何将整体的端到端ML管道重构为特征工程和模型训练管道。

2. 什么是MLOps

MLOps是最近出现的一个术语，描述了如何将DevOps原理应用于自动化ML系统的构建，测试和部署。持续交付基金会SIG-MLOps将MLOps定义为：“是DevOps方法论的扩展，将机器学习和数据科学资产作为DevOps生态中的一等公民”。MLOps旨在统一ML应用程序的开发和操作，使得团队可以更容易更频繁地部署更好的模型。Martinfowler.com将MLOps定义为：“一种软件工程方法，其中跨职能团队能基于代码、数据和模型以较小且安全的增量生成机器学习应用程序，并且可以在较短的周期内被复制和可靠地发布。”

与Devops相比，MLOps的一些主要挑战是如何处理版本化数据（不仅仅是版本化代码），如何管理专用硬件（GPU）以及如何管理模型的数据治理和合规性。

3. DevOps vs MLOps

Git是世界上最受欢迎的源代码版本控制系统，它用于跟踪随时间的变化并支持不同版本的源代码。支持版本控制是自动化和持续集成（CI）解决方案的先决条件，因为它可以以完全自动化的方式对任何环境进行可复制的配置。也就是说我们假定环境所需的配置信息和将要测试系统的源代码都存储在版本控制系统中。通常，在使用DevOps时，每次Git提交都会触发软件包的自动创建，这些软件包可以仅使用版本控制中的信息就可以部署到任何环境中。

在大多数DevOps配置中，一般Jenkins与Git一起用作自动化服务器，以可控制、可预测的方式构建、测试和部署版本化代码。Jenkins对于CI / CD管道遵循的典型步骤是：提供测试虚拟机（VM）/容器，将代码签出到计算机上，编译代码，运行测试，打包二进制文件和部署二进制文件。对于 Java 而言，在将二进制文件部署至暂存或生产系统中之前，会运行诸如maven之类的构建 工具 来编译、测试和打包二进制文件。对于Docker，这意味着编译Dockerfile并将 Docker 镜像部署到Docker注册表。

MLOps最具代表性的特征可能是需要对数据和代码进行版本控制，以实现可重现的训练模型工作流。Git不适合作为控制数据版本的平台，因为它无法扩展以存储大量数据。

对于MLOps而言，Git和Jenkins还远远不够，因为MLOps的构建过程需要运行复杂的分布式工作流，同时我们需要带版本的代码和数据，以确保可重现自动化构建。工作流就是我们所说的ML管道，即组件图，其中每个组件都有入参和数据，成功的工作流会将模型部署到生产中。标准ML管道至少包括以下组件：验证输入数据，计算输入数据的特征，生成训练/测试数据，训练模型，验证模型，部署模型以及在生产中监视模型。实际生产环境中这种简化的管道可能会更加复杂，其中模型训练阶段可以细分为超参数调整、 模型简化测试和分布式训练。

已经有许多支持运行业务流程ML管道的端到端ML框架：TensorFlow Extended（TFX）支持Airflow、Beam和Kubeflow管道；Hopsworks支持Airflow；MLFlow支持Spark；Kubeflow支持Kubeflow管道。这些框架使工作流能够自动执行，并且可重复执行，例如仅更改输入参数就可以重新训练模型，具有在组件之间传递数据的能力以及指定基于事件触发工作流的能力（例如 在一天的特定时间，新数据到达时或模型性能降到给定水平以下时）。TFX，MLFlow和Hopsworks还支持使用Beam或Spark进行分布式处理，从而支持在使用大量数据的集群上横向扩展。

3. MLOps: 代码和数据版本化

3.1 Git风格的数据版本

由Dmitry Petrov开发的DVC，提供了一种对云存储中的文件/对象进行版本控制的开源工具，该工具使用Git来存储有关文件和reflink（支持数据文件的透明写时复制）的元数据，以确保 git目录和数据文件的一致性。类似地，Kubernetes上的ML平台Pachyderm也提供了使用类似git语义的数据版本控制平台。但是，这些类似git的方法只跟踪不可变的文件，而不存储文件之间的差异。它们无法处理时间旅行（time-travel）查询，例如“给我2016/2018年范围内的训练/测试数据”或“给我这些特征在2018年9月6日的价值”。如果没有时间旅行，它们将无法支持增量特征工程，如仅对自上次运行（1小时前，一天前等）以来发生变化的数据计算特征。

3.2 时间旅行查询和增量拉取的数据版本控制

类似于git的数据版本控制系统的替代方法是使用提供版本化、结构化数据集的事务数据湖。版本化的数据集不仅具有其数据的模式（schema）版本，其中模式可能会随着时间而演化，而且对数据湖的更新是原子化的，并通过提交（commit）进行标识。最著名的此类平台是开源项目：Delta Lake，Apache Hudi，Apache Iceberg。用户可以执行时间旅行查询以返回给定的时间点（commit-id）的数据，或者返回给定时间间隔的数据，或者从给定的时间点变更的数据。它们使用索引（ bloom filters, z-indexes, data-skipping indexes）高效地执行时间旅行查询，这些索引大大减少了需要从文件系统或对象存储中读取的数据量。事务性数据湖还允许客户端仅读取给定时间点以来数据集中的变更，从而可以开启增量特征工程，即仅针对最近一小时或一天中变更的数据计算特征。

4. Hopsworks特征存储

用于机器学习的特征存储是一种特征计算和存储服务，它使特征可以被注册、发现和用作ML管道的一部分以及用于模型推理的在线应用程序。通常需要特征存储来存储大量特征数据并为在线应用程序提供对特征的低延迟访问。它们通常实现为双数据库系统：低延迟在线特征存储（通常是键值存储或实时数据库）和横向扩展 SQL 数据库，用于存储大量特征数据，用于训练和批处理应用程序。在线特征存储使在线应用程序能够在执行推理请求之前以接近实时的特征数据丰富特征向量。离线特征存储可以存储大量特征数据，这些特征数据用于创建训练/测试数据以用于模型开发，或者用于批处理应用程序以用于模型评分。特征存储解决了ML管道中的以下问题：

• 通过在团队/项目之间共享特征以复用特征管道；

• 能够大规模且低延迟地提供特征；

• 确保训练和服务之间特征的一致性，一次特征工程后便可以缓存在在线和离线特征存储中；

• 确保特征在不同时间点的正确性，在做出预测并在稍后获取结果时，也需要能够查询过去给定时间点上不同特征的值。

ML的特征存储由在线和离线数据库组成，并将来自后端系统的原始数据转换为经过设计的特征，这些特征可供在线和批处理应用程序进行推理，并可供数据科学家创建用于模型开发的训练/测试数据。

大多数大型AI公司（Uber，Twitter，AirBnb，Google，Facebook，Netflix，Comcast）都建立了自己内部特征库，但也有两个开源特征库：Hopsworks特征库（基于Apache Hudi / Hive，MySQL Cluster和 HopsFS）和Feast （基于Big Query，BigTable和 Redis 构建）。特征存储还使用的其他数据库包括Cassandra，S3和Kafka，以及自定义键值存储。

4.1. Hopsworks特征存储的端到端ML管道

MLOps和DataOps CI/CD管道与传统DevOps的不同之处在于，它们可能由新的数据到达时进行处理而触发（以及由于数据工程或模型训练管道的源代码更新而触发）。DataOps涉及数据处理管道（在我们的情况下为特征管道）的自动化测试和部署，以及数据验证和数据管道等阶段。另外MLOps还涉及自动化训练和部署ML模型，以及模型训练、模型验证和模型部署等阶段。

特征存储支持将ML工作流分解为两个工作流：（1）用于工程特征的“DataOps”工作流，并验证将特征存储在特征存储的数据，以及（2）用于训练模型的“ MLOps”工作流，使用特征存储中的特征，分析和验证这些模型，将其部署到在线模型服务基础架构中以及监视生产中的模型性能。

一些ML生命周期框架（例如TensorFlow Extended（TFX）和MLFlow），都是基于端到端ML管道，这些管道以原始数据开始并以生产模型结束。但是，端到端ML管道的第一步将原始数据转换为模型的训练数据可能会非常昂贵。Airbnb报告称如果没有特征存储，创建训练/测试数据可能会花费数据科学家多达60-80％的时间。特征存储使转换后的数据（特征）可以在不同模型中复用。有了特征存储后，不再需要从原始数据到模型的端到端ML管道。可以将端到端ML管道分解为两个单独的管道，每个管道都以自己的节奏运行：（1）特征管道，这些数据管道从后端系统中提取数据，对其进行验证，特征化并缓存在特征存储中；以及（2 ）训练管道，该训练管道从特征数据训练模型，验证那些模型并将其部署到生产中。

引入用于MLOps的特征存储的动机是，提取和特征化新数据的过程与使用许多不同来源的特征的训练模型的过程是分开的。也就是说，与模型训练的节奏相比，特征工程的节奏通常存在差异。有些特征可能每隔几秒钟更新一次，而其他特征则每隔几个月更新一次。另一方面，可以按需（定期（例如每天或每周））或在监视显示模型的性能下降时对模型进行训练。当新数据到达时，特征工程流水线通常以固定的间隔触发；当将源代码推送到git时，特征工程流水线通常按需触发，因为变更了特征的设计方式。

4.2. 有状态的ML管道

开发数据管道的最佳实践是使它们无状态且幂等的，以便在发生故障时可以安全地重新运行它们。但是，ML管道是具有状态的。在将模型部署到生产之前，你需要一些上下文信息：该模型的性能是否比当前部署的模型好？该决策需要有关当前部署模型的状态信息。理想情况下，我们还需要历史状态，这样我们可以随时间观察和评估模型的性能，以及随时间推移构建模型的处理时间/成功率。Hopsworks、TFX和MLFlow提供了一个元数据存储，以使ML管道能够做出有状态的决策，记录其执行步骤，存储它们产生的artifacts以及存储最终模型的来源。TFX和MLFlow都很麻烦，开发人员使用其组件模型（每个阶段都有明确定义的输入和输出）在每个阶段都需要重写代码，这样他们可以截取组件的输入参数，并将它们记录到元数据存储中。Hopsworks提供了一个很好的元数据模型，在该模型中，管道可以对HopsFS（HDFS）文件系统进行读/写操作，并使用Hopsworks API与特征存储进行交互。这样，元数据事件、artifacts、执行（execution）和出处就隐式存储到元数据存储中，而无需像TFX或MLFlow那样重写notebook或 python 程序。

5. 特征管道反馈Hopsworks特征存储

特征存储使特征管道能够缓存特征数据以供许多下游模型训练管线使用，从而减少了创建/回填特征的时间。特征组通常一起计算，并具有自己的摄取节奏，请参见上图。可以使用流应用程序每隔几秒钟实时更新在线特征存储中的特征，而批处理特征可以每小时，每天，每周或每月更新。

在实践中，特征管道是数据管道，该管道的输出是经过清理、验证和特征化的数据。由于通常无法保证输入数据的正确性，因此必须验证输入数据，并且必须处理所有丢失的值（通常通过估算或忽略它们）。TFX数据验证和AWS Deequ是两种流行的数据验证框架，它们支持扩展传统的基于模式的数据验证（例如，此列包含整数）以及数据验证规则，以检查数值或分类值是否等于预期。例如，虽然架构确保数值特征为浮点类型，但还需要其他验证规则以确保这些浮点在预期范围内。还可以进一步检查以确保列的值是唯一的，而不是null，以确保其描述性统计信息在一定范围内。然后，将经过验证的数据转换为数字和分类特征，然后将其缓存在特征存储中，随后将其用于训练模型以及进行批处理/在线模型推断。

特征管道与数据管道共享许多相同的最佳实践DevOps实践。数据/特征自动测试的类型包括：

• 所有特性代码的单元测试和集成测试（将代码推送到Git时，Jenkins可以运行这些测试）；

• 测试特征值是否在预期范围内（TFX数据验证或Deequ）；

• 测试特征的唯一性，完整性和独特性（Deequ）；

• 测试特征分布是否符合预期（TFX数据验证或Deequ）；

• 测试每个特征与标签之间的关系，以及各个信号之间的成对相关性（Deequ）；

• 测试每个特征的成本（自定义测试）；

• 测试个人信息没有泄漏到特征中（自定义测试）。

当特征存储可用时，特征流水线的输出就是缓存特征数据并存储到特征存储。理想情况下，目标数据输出需要支持版本化数据，例如Hopsworks特征存储中的 Apache Hudi 。在Hopsworks中，特征流水线将数据向上插入（插入或更新）到现有特征组中，其中特征组是一起计算的一组特征（通常是因为它们来自同一后端系统，并且由某些实体或键关联）。每当运行特征管道时，都会在Hudi数据集中创建一个新的提交。这样我们可以跟踪和查询对特征存储中特征组的不同提交，并监视随时间变化的摄取数据统计信息的变化。

6. 从特征存储开始的模型训练管道

模型训练管道属于MLOps范式，在该模型中，从Hopsworks特征存储中的Apache Hudi读取版本化的特征，以创建训练/测试数据，用于训练模型，然后在生产中对其进行部署和监视。ML artifacts和执行（execution）的来源存储在Hopsworks的元数据存储中，并且ML管道由Hopsworks协调。

使用特征存储进行模型训练通常在工作流中涉及至少三个阶段（或程序）：

• 选择特征，文件格式以及用于从特征存储中的特征创建的训练/测试数据集的文件系统（或对象存储）。注意，对于Hopsworks特征存储，还可以提供时间戳（对应于Hudi commit-id）来重现训练/测试数据集，就像过去的某个时间点一样。

• 使用在步骤1中创建的训练数据集来训练模型（训练可以进一步分解为以下步骤：超参数优化，模型简化测试和模型训练）；

• 使用自动化测试验证模型，并将其部署到批处理应用程序的模型注册表和/或在线应用程序的在线模型服务器。

在Hopsworks平台中，这三个步骤通常是python程序或Jupyter notebooks，它们作为Airflow DAG（有向无环图）的一部分执行。也就是说，Airflow协调了管道的执行。Airflow使DAG可以定期进行调度，但是也可以配置为在新特征数据到达特征存储区或模型训练管道代码推送Git提交时运行工作流。

在模型验证步骤中执行的自动测试的类型包括：

• 测试模型如何在不同的数据切片上执行以检查偏差。

• 测试模型对分布特征向量的鲁棒性。

Hopsworks支持Jupyter notebook使用Google的假设分析（What-if）工具进行模型分析。研究反事实（将数据点与模型预测不同结果的最相似点进行比较）时非常有用，这样可以更轻松地开发之后在生产管道中使用的模型验证测试。

Google的假设分析工具可用于分析模型，询问反事实并测试不同数据片段上的偏差。此处的知识发现可以转移到模型验证测试中。

6.1 监控在线模型

将模型部署到模型服务器以供在线应用程序使用时，我们需要监视模型的性能及其输入特征。我们需要确定生产中的输入特征在统计上是否不同于用于训练模型的输入特征。在实践中，我们可以通过将在训练数据（可通过特征存储API调用访问）上计算出的统计数据与在运行时从输入特征中收集的统计数据进行比较来做到这一点。在Hopsworks中，我们会将模型的所有预测请求发送到Kafka中的主题。然后便可以编写一个Spark Streaming或Flink应用程序，该应用程序在Kafka中处理预测请求，在基于时间的窗口中计算统计信息，并将这些统计信息与特征存储中的训练数据统计信息进行比较。如果给定特征基于时间的Windows统计信息与训练统计信息相差很大，则流应用程序可以通知ML工程师输入功能与预期不符，流应用程序通常还可以为模型计算业务级别的KPI，并提供一个UI，以使操作员能够可视化模型的性能。更具体地说，要在在线监视中查找的错误信号包括：

概念漂移（Concept drift）

在模型中，目标变量是模型试图预测的变量。例如，可能是金融交易被怀疑是欺诈或不是欺诈。当模型的统计属性以非预期的方式随时间变化时（例如出现了一个新的欺诈方案，该欺诈方案增加了欺诈的总量），概念就会漂移。

数据漂移（Data drift）

如果输入特征的统计属性以意外的方式随时间变化，则会对模型的性能产生负面影响。例如，如果用户由于假期而执行了比正常情况多得多的金融交易，但模型并未经过训练以处理假日，则模型的性能可能会降低（丢失欺诈行为或将太多交易标记为可疑） 。

特征管道变化（Feature pipeline changes）

如果在特征管道中计算特征的方式发生了变化，并且在线模型使用在线特征存储中的特征数据来丰富其特征向量，则可能会对模型的性能产生负面影响。例如，如果更改了计算用户执行的交易数量的方式，则可能会对模型的性能产生负面影响。

7. 总结

现在我们已经基于MLOps原理的特征存储涵盖了端到端ML管道。通过更新管道代码或新到达的数据，可以对变更进行持续测试，并可以持续更新模型并将其部署到生产环境中。我们展示了特征存储如何使整体式端到端ML管道分解为特征管道和模型训练管道。我们还讨论了如何使用现代数据湖框架（如Apache Hudi）进行数据版本控制。在下一个博客我们将更详细地介绍ML管道和可重复的Hopsworks实验，以及如何轻松地将管道从开发环境转移到生产环境，我们还将展示如何使用Airflow开发功能管道和模型训练管道。