本章内容包括：

• 定义工作流和工作流编排
• 深度学习系统为何需要支持工作流
• 设计通用的工作流编排系统
• 介绍三个开源的编排系统：Airflow、Argo Workflows和Metaflow

在本章中，我们将讨论深度学习系统中最后但至关重要的一部分：工作流编排。工作流编排是一种管理、执行和监控工作流自动化的服务。工作流是一个抽象且广泛的概念，它实质上是一系列操作的序列，这些操作是某个更大任务的一部分。如果您可以制定一套任务来完成一项工作，那么这个计划就是一个工作流。例如，我们可以为训练机器学习（ML）模型定义一个顺序工作流。这个工作流可以由以下任务组成：获取原始数据、重建训练数据集、训练模型、评估模型和部署模型。

因为工作流是一个执行计划，它可以手动执行。例如，数据科学家可以手动完成我们刚刚描述的模型训练工作流中的任务。例如，要完成“获取原始数据”任务，数据科学家可以编写网络请求并发送给数据集管理（DM）服务来获取数据集，所有这些都不需要工程师的帮助。

然而，手动执行工作流并不理想。我们希望自动化工作流的执行。当有许多为不同目的开发的工作流时，我们需要一个专门的系统来处理工作流执行的复杂性。我们将这种系统称为工作流编排系统。

工作流编排系统被设计用于管理工作流的生命周期，包括工作流的创建、执行和故障排除。它不仅提供了保持所有计划代码运行的脉搏，还提供了一个控制平面，供数据科学家管理深度学习系统中的所有自动化操作。

在本章中，我们将讨论工作流编排系统的设计以及在深度学习领域中使用最广泛的开源编排系统。通过阅读本章，您不仅将对系统需求和设计选项有扎实的理解，还将了解如何选择最适合您自己情况的合适的开源编排系统。

介绍工作流编排

在我们深入讨论工作流编排系统设计的细节之前，让我们先简要讨论一下工作流编排的基本概念，特别是从深度学习/机器学习的角度看工作流的特殊挑战。

注意，因为在深度学习项目和机器学习项目中使用工作流编排的要求几乎是相同的，所以在本章中，我们将深度学习和机器学习这两个词互换使用。

什么是工作流？

一般来说，工作流是一个由一系列操作组成的任务。工作流可以被看作是一个有向无环图（DAG），其中每个步骤是最小的可恢复计算单元，描述了一个动作，例如获取数据或触发一个服务。每个步骤要么成功完成，要么完全失败。在本章中，我们将任务（task）和步骤（step）互换使用。

DAG指定了步骤之间的依赖关系和执行顺序。图9.1显示了一个用于训练自然语言处理（NLP）模型的示例工作流。

从图9.1中的示例DAG可以看出，工作流由许多步骤组成。每个步骤依赖于另一个步骤，实线箭头表示步骤之间的依赖关系。这些箭头和步骤形成了一个没有循环的工作流DAG。

如果按照DAG中的箭头（从左到右）进行操作并完成任务，您可以训练并发布一个NLP模型到生产环境中。例如，当一个传入请求触发工作流时，授权（authorization）步骤将首先被执行，然后同时执行数据集构建步骤和嵌入获取步骤。箭头的另一侧的步骤将在这两个步骤完成后执行。

工作流在IT行业中被广泛使用。只要您可以将一个过程定义为一系列单个任务/步骤的DAG，这个过程就可以被认为是一个工作流。工作流对于深度学习模型开发至关重要。实际上，在生产环境中，大多数深度学习模型构建活动都以工作流的形式呈现和执行。

注意：工作流不应该有循环。为了确保工作流能够在任何情况下完成，其执行图必须是一个DAG，以防止工作流执行陷入死循环。

什么是工作流编排？

一旦我们定义了一个工作流，下一步就是运行工作流。运行工作流意味着根据工作流的DAG中定义的顺序执行工作流步骤。工作流编排是我们用来描述工作流的执行和监控的术语。

工作流编排的目标是自动化执行工作流中定义的任务。在实践中，工作流编排的概念通常扩展到整个工作流管理领域，包括以自动化的方式创建、调度、执行和监控多个工作流。

为什么深度学习系统需要工作流编排？理想情况下，我们应该能够将整个深度学习项目编写为一个整体。这正是我们在项目的原型设计阶段所做的，将所有的代码放在一个Jupyter笔记本中。那么，为什么我们需要将原型设计的代码转换为工作流，并在工作流编排系统中运行呢？答案有两个方面：自动化和工作共享。为了理解这些原因，让我们来看一下图9.2中的三个示例训练工作流。

使用工作流的一个巨大好处是将大块的代码转化为一组可共享和可重用的组件。在图9.2中，我们设想了三名数据科学家分别在三个模型训练项目（A、B和C）上工作。由于每个项目的训练逻辑不同，数据科学家开发了三个不同的工作流程（A、B和C）来自动化他们的模型训练过程。尽管每个工作流程具有不同的DAG，但每个DAG中的步骤高度重叠。总共的六个步骤是可共享和可重用的。例如，auth步骤（步骤1）是所有三个工作流程的第一步。

拥有可重用的步骤可以极大提高数据科学家的生产力。例如，要从DM服务中提取数据（图9.2中的步骤2），数据科学家需要学习DM网络API的工作方式。但是，如果已经有人将DM数据提取方法构建为一个步骤函数，科学家们可以在他们的工作流程中直接重用这个步骤，而无需学习如何与DM服务进行交互。如果每个人都以工作流的形式编写项目，我们将拥有许多可重用的步骤，这将在组织层面上节省大量重复的工作！

工作流之所以适用于深度学习开发的另一个原因是它促进了协作。模型开发需要团队合作；一个专门的团队可能负责数据，而另一个团队负责训练算法。通过在工作流程中定义一个复杂的模型构建过程，我们可以将一个复杂的大型项目分解为不同的部分（或步骤），并将它们分配给不同的团队，同时保持项目有序和组件的正确顺序。工作流DAG清楚地显示了所有项目参与者所见的任务依赖关系。

简而言之，一个好的工作流编排系统鼓励工作共享，促进团队协作，并自动化复杂的开发场景。所有这些优点使得工作流编排成为深度学习项目开发的关键组成部分。

深度学习中使用工作流编排面临的挑战

在前面的部分中，我们看到工作流系统如何为深度学习项目开发提供了许多好处。但是有一个注意事项：使用工作流来原型化深度学习算法的想法是繁琐的。

为了了解为什么会繁琐以及为什么会繁琐，让我们看一下深度学习开发过程的图表（图9.3）。这个图表应该为你理解工作流在深度学习环境中带来的挑战奠定基础。

在图9.3中，我们可以看到一个从数据科学家的角度来看的典型深度学习项目开发过程。该过程可以分为两个阶段：本地孵化阶段和生产阶段。

在本地孵化阶段，数据科学家在他们的本地/开发环境中进行数据探索和模型训练的原型制作。当原型制作完成并且项目看起来有希望时，数据科学家开始进行生产部署：将原型代码移植到生产系统中。 在生产阶段，数据科学家将原型代码转换为工作流。他们将代码分解为多个步骤，并定义一个工作流DAG，然后将工作流提交给工作流编排系统。之后，编排系统接管并根据其计划运行工作流。

原型制作和生产之间的差距

如果你问一个从事工作流编排系统的工程师对图9.3中的开发过程的看法，答案很可能是：还不错！但在实践中，这个过程对数据科学家来说是有问题的。

从数据科学家的角度来看，一旦算法在本地测试通过，其原型代码应该立即交付到生产环境中。但在图9.3中，我们可以看到原型制作阶段和生产阶段之间并没有平稳的连接。将孵化代码交付到生产环境并不直观；数据科学家必须额外工作来构建一个工作流，在生产环境中运行他们的代码。原型代码与生产工作流之间的差距对开发速度有两个方面的影响：

• 工作流的构建和调试并不直观：数据科学家在编写模型训练工作流时通常面临巨大的学习曲线，需要学习工作流DAG语法、工作流库、编码范式和故障排查等知识。工作流的故障排查是最令人痛苦的部分。大多数编排系统不支持本地执行，这意味着数据科学家必须在远程编排系统中测试他们的工作流。这很困难，因为工作流环境和工作流执行日志都是远程的，所以当工作流执行出错时，数据科学家不能轻易地找出根本原因。
• 工作流的构建不仅仅是一次性的：人们常常错误地认为，工作流的构建只发生一次，所以即使耗时费力也没关系。但事实是，工作流的构建是持续发生的，因为深度学习开发是一个迭代的过程。正如图9.3所示，数据科学家不断地进行原型制作和生产实验的迭代，因此工作流需要经常更新，以测试从本地到生产环境的新改进。因此，令人不愉快且耗时的工作流构建会反复发生，阻碍了开发速度。

平滑过渡从原型制作到生产

尽管存在差距，图9.3中的流程是很好的。数据科学家从一个直观的脚本开始进行原型制作，然后继续工作。如果每次迭代后的结果足够有希望，那么“直观的本地脚本”将被转换为工作流，并在生产环境中运行。

关键的改进在于使原型代码到生产工作流的过渡步骤无缝连接。如果一个编排系统专为深度学习场景设计，它应该提供工具来帮助数据科学家以最小的工作量从他们的代码构建工作流。例如，Metaflow是一个开源库，将在9.3.3节中讨论，它允许数据科学家通过编写带有Python注解的Python代码来授权工作流。数据科学家可以直接从他们的原型代码中获得一个工作流，而无需进行任何更改。Metaflow还在本地和云生产环境之间提供统一的模型执行用户体验。这消除了工作流测试中的摩擦，因为Metaflow在本地和生产环境中以相同的方式操作工作流。

设计一个流程编排系统

在本节中，我们将分三个步骤来设计工作流编排系统。首先，我们将使用一个典型的数据科学家用户场景，展示编排系统从用户角度的工作方式。其次，我们学习通用的编排系统设计。第三，我们总结构建或评估编排系统的关键设计原则。通过阅读本节内容，您将了解编排系统的一般工作原理，从而可以自信地评估或使用任何编排系统。

用户场景

虽然工作流程的过程在不同的场景下有很大的差异，但对于数据科学家来说，用户场景是相当标准的。大多数工作流使用可以分为两个阶段：开发阶段和执行阶段。请参考图9.4，了解数据科学家Vena的工作流用户体验。让我们一步一步地跟随图9.4中Vena的用户场景进行说明。

开发阶段

在开发阶段，数据科学家将他们的训练代码转化为工作流。以下是Vena的示例：

1. Vena是一位数据科学家，她在本地环境中使用Jupyter笔记本或纯Python原型化她的模型训练算法。经过本地测试和评估后，Vena认为现在是将代码部署到生产环境中进行实时实验的时候了。
2. 因为在生产环境中运行的所有内容都是工作流，所以Vena需要将她的原型代码转化为工作流。因此，Vena使用工作流编排系统提供的语法将她的工作重建为一个任务DAG，并将其配置在一个YAML（文本配置）文件中。例如，数据解析 -> 数据增强 -> 数据集构建 -> 训练 -> [在线评估，离线评估] -> 模型发布。
3. Vena为DAG中的每个步骤设置输入/输出参数和操作。以训练步骤为例，Vena将步骤操作设置为RESTful HTTP请求。该步骤将向模型训练服务发送一个RESTful请求来启动一个训练作业。此请求的有效载荷和参数来自于步骤的输入参数。
4. 工作流定义完成后，Vena在DAG YAML文件中设置工作流的执行计划。例如，Vena可以将工作流安排在每个月的第一天运行，并设置工作流由外部事件触发。
5. Vena进行本地验证并提交工作流给编排服务。

为了让您了解工作流在实际中的含义，以下代码展示了Vena的一个伪代码工作流（在第9.3节中，我们将讨论实际的工作流系统）：

# define workflow DAG
with DAG(
  description='Vena’s sample training workflow',
  schedule_interval=timedelta(months=1),
  start_date=datetime(2022, 1, 1),
  ) as dag:
  
  # define execution logic for each step
  data_parse_step = BashOperator( .. .. ..)
  data_augment_step = BashOperator( .. .. ..)
  dataset_building_step = BashOperator( .. .. ..)
  training_step = BashOperator( .. .. ..)
  
  # Declares step dependencies
  data_parse_step >> data_augment_step
  >> dataset_building_step >> training_step

执行阶段

在执行阶段，编排服务执行模型训练工作流，就像Vena的示例一样：

1. 一旦Vena的工作流被提交，编排服务将工作流DAG保存到数据库中。
2. 编排服务的调度器组件检测到Vena的工作流，并将工作流的任务分派给后台工作节点。调度器将确保任务按照工作流DAG中定义的顺序执行。
3. Vena使用编排服务的Web界面实时检查工作流的执行进度和结果。
4. 如果工作流生成了一个好的模型，Vena可以将其推广到预发布和生产环境。如果不是，则Vena开始另一次原型化迭代。

判断一个编排系统是否适用于深度学习的关键指标是将原型化代码转化为工作流的难易程度。在图9.4中，我们看到每当Vena原型化一个新的想法时，她都需要将训练代码转化为工作流。如果我们能减少将深度学习代码转化为工作流的难度，可以想象节省的人力时间将会有多大。

注意：工作流应该始终保持轻量级。工作流用于自动化一个过程，其目标是将一系列任务进行分组和连接，并按照定义的顺序执行。使用工作流的重要好处是人们可以共享和重用任务，从而更快地自动化他们的过程。因此，工作流本身不应进行任何繁重的计算，真正的工作应由工作流中的任务完成。

一个通用编排系统设计

现在让我们转向通用的工作流编排系统。为了帮助您理解编排系统的工作原理并研究开源编排系统，我们准备了一个高级系统设计。通过放大详细实现细节并仅保留核心组件，这个设计适用于大多数编排系统，包括将在第9.3节中讨论的开源系统。请参考图9.5的设计提案。

一个工作流编排系统通常包含以下五个组件：

• Web服务器：Web服务器提供Web用户界面和一组Web API，供用户创建、检查、触发和调试工作流的行为。
• 调度器和控制器：调度器和控制器组件有两个功能。首先，调度器监视系统中的每个活动工作流，并在适当的时候安排工作流运行。其次，控制器将工作流任务分派给工作节点。虽然调度器和控制器是两个不同的功能单元，但它们通常一起实现，因为它们都与工作流的执行相关。
• 元数据数据库：元数据数据库存储工作流的配置、DAG、编辑和执行历史，以及任务的执行状态。
• 工作节点组：工作节点组提供计算资源来运行工作流任务。工作节点抽象了基础设施，并且对正在运行的任务是无关的。例如，我们可能有不同类型的工作节点，比如Kubernetes工作节点和亚马逊弹性计算云（EC2）工作节点，但它们都可以执行相同的任务，尽管在不同的基础设施上运行。
• 对象存储：对象存储是所有其他组件共享的文件存储；它通常建立在云对象存储（例如亚马逊简单存储服务S3）之上。对象存储的一个用途是任务输出共享。当一个工作节点运行一个任务时，它从对象存储中读取上一个任务的输出值作为任务的输入；工作节点还将任务的输出保存到对象存储中，以供后续任务使用。

对象存储和元数据数据库对编排系统的所有组件都是可访问的，包括调度器、Web服务器和工作节点的组件。通过集中存储数据，核心组件可以解耦，使得Web服务器、调度器和工作节点可以独立工作。

工作流是怎么执行的？

首先，Vena定义工作流的DAG。在DAG内部，Vena声明了一组任务并定义了任务执行顺序的控制流程。对于每个任务，Vena可以使用系统的默认运算符（例如Shell命令运算符或Python运算符），或者构建自己的运算符来执行任务。

其次，Vena通过Web界面或命令行将工作流（包括依赖代码）提交给Web服务器。工作流被保存在元数据数据库中。

第三，调度器定期（每几秒或几分钟）扫描元数据数据库，并检测新的工作流；然后在预定的时间启动工作流。为了执行工作流，调度器调用控制器组件，根据DAG中定义的任务序列，将工作流的任务分派到工作队列中。

第四步，工作节点从共享的作业队列中获取一个任务；它从元数据数据库中读取任务定义，并通过运行任务的操作符执行任务。在执行过程中，工作节点将任务的输出值保存到对象存储中，并将任务的执行状态报告给元数据数据库。

最后但同样重要的是，Vena使用托管在Web服务器组件上的Web用户界面来监视工作流的执行。由于调度器/控制器组件和工作节点实时向元数据数据库报告状态，因此Web用户界面始终显示最新的工作流状态。

工作流编排设计原则

因为我们已经了解了工作流编排系统的内部和外部工作原理，现在是时候审视使一个工作流编排系统在深度学习场景下脱颖而出的设计原则了。我们希望您可以将这些原则作为指导，用于改进您的系统或评估开源方法。

注意，从工程角度来看，工作流编排系统是深度学习系统中最复杂的组件之一，因此在最初的几个版本中，不必过于担心使您的系统与这些原则完全匹配。

原则1：关键性

工作流编排本质上是一个作业调度挑战，因此任何编排系统的底线都是提供稳定的工作流执行体验。有效的工作流应该能够正确、可重复地按计划执行。

原则2：易用性

在深度学习环境中，编排系统的易用性衡量标准在于它是否能提高数据科学家的生产力。大多数数据科学家与编排系统的交互都是创建、测试和监视工作流。因此，用户友好的编排系统应该让用户能够轻松地创建、监视和排查工作流。

原则3：可扩展性

为了满足各种各样的深度学习基础设施，人们应该能够轻松地定义自己的任务操作符和执行器，而不必担心它们部署在哪里。编排系统应该提供适合您环境的抽象级别，无论是Amazon EC2还是Kubernetes。

原则4：隔离性

关键的隔离性包括两种：工作流创建隔离和工作流执行隔离。工作流创建隔离意味着在创建工作流时，人们不能相互干扰。例如，如果Vena提交了一个无效的工作流DAG，或者发布了一个被其他工作流引用的常用共享库的新版本，现有的工作流不应受到影响。

原则5：可扩展性

一个优秀的编排系统应该解决以下两个可扩展性问题：处理大量并发的工作流和处理庞大的复杂工作流。并发工作流的扩展性通常意味着在足够的计算资源下，例如向工作组添加更多的工作节点，编排系统可以处理无限数量的并发工作流执行。同时，系统应始终保持每个工作流的服务级别协议（SLA）。例如，无论有多少其他工作流正在执行，工作流应该在其计划的时间内执行，而且不晚于2秒。

对于单个大型工作流的扩展性，系统应鼓励用户不用担心性能问题，以便专注于可读性好、简单明了的代码和易于操作。当工作流执行达到某个限制时（例如，训练操作符执行时间过长），编排系统应提供一些水平并行性操作符，例如分布式训练操作符，以解决单个工作流的性能问题。

深度学习编排的主要扩展思想是我们应该在系统级别解决性能问题，而不是要求用户编写具有可扩展性意识的代码。这样可以避免降低代码的可读性，增加调试难度并增加运维负担。

原则6：以人为中心支持原型和生产

将数据科学家的本地原型代码与生产工作流连接起来是深度学习特有的要求。这是我们评估编排系统是否适用于深度学习系统的关键指标。

为深度学习设计的编排系统将尊重从原型到生产的迭代式持续开发过程。因此，它将全力帮助数据科学家将本地原型代码无缝转换为生产工作流。

开源方案

在本节中，我们将介绍三个经过实战验证的工作流编排系统：Airflow、Argo Workflows和Metaflow。这三个开源系统在IT行业得到了广泛应用，并且有着活跃的社区支持。除了对它们进行一般性的介绍，我们还将从深度学习项目开发的角度评估这些工作流系统。

为了进行公正的比较，我们将在Airflow、Argo Workflows和Metaflow中实现相同工作流的伪代码。基本上，如果有新的数据，我们会首先对数据进行转换并将其保存到数据库中的新表中，然后通知数据科学团队。此外，我们希望工作流每天运行一次。

Airflow

Airflow（airflow.apache.org/docs/apache…）于2014年在Airbnb创建，并现在是Apache基金会的一部分。Airflow是一个用于以编程方式编写、调度和监控工作流的平台。Airflow并非专为深度学习场景而设计，它最初是为了编排日益复杂的ETL（提取、转换、加载）流水线（或数据流水线）而构建的。但由于Airflow具有良好的可扩展性、生产质量和图形用户界面（GUI）支持，它在许多其他领域，包括深度学习，都得到了广泛应用。截至本书撰写时，Airflow是最受欢迎的编排系统。

典型的使用案例是在Airflow中构建工作流需要两个步骤。首先，定义工作流的DAG和任务。其次，在DAG中声明任务之间的依赖关系。Airflow的DAG本质上是Python代码。请参考以下示例，了解我们的样例工作流在Airflow中的实现方式。

# declare the workflow DAG.
with DAG(dag_id="data_process_dag",
         schedule_interval="@daily",
         default_args=default_args, 
         template_searchpath=[f"{os.environ['AIRFLOW_HOME']}"],   
         catchup=False) as dag:

# define tasks of the workflow, each code section below is a task
is_new_data_available = FileSensor(
    task_id="is_new_data_available",
    fs_conn_id="data_path",
    filepath="data.csv",
    .. .. .. 
)

# define data transformation task
transform_data = PythonOperator(
    task_id="transform_data",
    python_callable=transform_data

)

# define table creation task
create_table = PostgresOperator(
    task_id="create_table",
    sql='''CREATE TABLE IF NOT EXISTS invoices (
            .. .. ..
            );''',
    postgres_conn_id='postgres',
    database='customer_data'
)

save_into_db = PythonOperator(
    task_id='save_into_db',
    python_callable=store_in_db

)

notify_data_science_team = SlackWebhookOperator(
    task_id='notify_data_science_team',
    http_conn_id='slack_conn',
    webhook_token=slack_token,
    message="Data Science Notification n"
    .. .. .. 
)

# Step two, declare task dependencies in the workflow
  is_new_data_available >> transform_data
  transform_data >> create_table >> save_into_db
  save_into_db >> notify_data_science_team
  save_into_db >> create_report

# The actual data transformation logic, which is referenced 
# in the “transform_data” task.
def transform_data(*args, **kwargs):
    .. .. ..

在代码清单9.1中，我们可以看到样例工作流DAG包含多个任务，例如create_table和save_into_db。在Airflow中，任务被实现为操作符（operator）。Airflow提供了许多预定义的和由社区管理的操作符，例如MySqlOperator、SimpleHttpOperator和DockerOperator。

Airflow的预定义操作符帮助用户在无需编码的情况下实现任务。您还可以使用PythonOperator来运行自定义的Python函数。一旦工作流DAG构建完成并且所有代码部署到Airflow中，我们可以使用UI或以下CLI命令来检查工作流的执行状态。以下是一些示例Shell命令：

airflow dags list

airflow tasks list data_process_dag

airflow tasks list data_process_dag --tree

如果您想了解更多关于Airflow的信息，可以查看其架构概述文档和教程（mng.bz/Blpw）。

关键功能

Airflow提供以下关键功能：

• DAGs：Airflow通过使用DAGs（有向无环图）来抽象复杂的工作流，工作流DAG是通过Python库来实现的。
• 编程式工作流管理：Airflow支持动态创建任务，并允许创建复杂的动态工作流。
• 出色的内置操作符以帮助构建自动化：Airflow提供了许多预定义的操作符，可以帮助用户在不编写代码的情况下完成任务。
• 坚实的任务依赖和执行管理：Airflow在每个任务中内置了自动重试策略，并提供了不同类型的传感器来处理运行时的依赖关系，例如检测任务完成、工作流运行状态变化和文件存在等。
• 可扩展性：Airflow使其传感器、钩子和操作符完全可扩展，这使得它能够从大量的社区贡献的操作符中受益。通过添加自定义操作符，Airflow也可以轻松集成到不同的系统中。
• 监控和管理界面：Airflow提供了强大的用户界面，用户可以快速查看工作流/任务的执行状态和历史。用户还可以从界面上触发和清除任务或工作流运行。
• 用于生产环境的高质量：Airflow提供了许多有用的工具，用于在生产环境中维护服务，如任务日志搜索、扩展性、警报和RESTful API。

局限性

• 对于数据科学家来说，使用Airflow需要付出较高的学习成本：Airflow对于实现不受内置操作符支持的任务有较高的学习曲线。此外，没有简单的方法来进行工作流的本地测试。

• 将深度学习原型代码转移到生产环境时存在较高的阻力：在将Airflow应用于深度学习时，数据科学家必须将他们的本地模型训练代码转换为Airflow DAG。这是额外的工作量，对于数据科学家来说是一种不愉快的体验，尤其是考虑到如果我们直接从模型训练代码构建工作流DAG的话，可以避免这种情况。

• 在Kubernetes上操作时存在较高的复杂性：在Kubernetes上部署和操作Airflow并不直观。如果您希望采用一个在Kubernetes上运行的编排系统，Argo Workflows是更好的选择。

Argo Workflows

Argo Workflows是一个开源的、基于容器的工作流引擎，用于在Kubernetes上编排并行工作流/任务。Argo Workflows解决了与Airflow相同的问题，但采用了不同的方式，它采用了与Kubernetes本地化的方法。

Argo Workflows与Airflow之间最大的区别在于，Argo Workflows是基于Kubernetes原生构建的。具体而言，Argo Workflows中的工作流和任务是以Kubernetes自定义资源定义（CRD）对象的形式实现的，每个任务（步骤）作为一个Kubernetes pod执行。请参见图9.6，了解高级系统概述。

在图9.6中，数据科学家Vena首先将工作流和其步骤/任务定义为Kubernetes CRD对象，通常以YAML文件的形式呈现。然后，她将工作流提交给Argo Workflows，其控制器在Kubernetes集群内创建CRD对象。接下来，Kubernetes动态启动Pod来按照工作流序列运行步骤/任务。 您可能还注意到，每个步骤的执行都通过容器和Pod完全隔离；每个步骤使用文件来表示其输入和输出值。Argo Workflows会自动将依赖文件挂载到步骤的容器中。

由Kubernetes Pod创建的任务隔离是Argo Workflows的重要优势。同时，简单性也是人们选择Argo Workflows的另一个原因。如果您了解Kubernetes，Argo的安装和故障排除都是直接的。我们可以使用Argo Workflows命令或标准的Kubernetes CLI命令来调试系统。

典型用例

为了更好地理解，让我们看一个Argo Workflows的示例。在本节中，我们使用Argo Workflows来自动化前面在Airflow部分中看到的相同的数据处理工作。工作流包括首先检查新数据，对数据进行转换，将其保存到数据库中的新表中，然后通过Slack通知数据科学家团队。请参见以下代码示例，其中定义了Argo Workflows的工作流。

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
 generateName: data-processing-
spec:
entrypoint: argo-steps-workflow-example
templates:
  - name: argo-steps-workflow-example
    Steps:
      - - name: check-new-data
          template: data-checker
      - - name: transform-data
          template: data-converter
          arguments:
            artifacts:
              - name: data-paths
                from: "{{steps.check-new-data.outputs.
                      artifacts.new-data-paths}}"

     - - name:save-into-db
         template: postgres-operator

     - - name: notify-data-science-team
         template: slack-messenger
         
 - name: data-checker
   container:
     image: docker/data-checker:latest
     command: [scan, /datastore/ds/]
   outputs:
     artifacts:
       - name: new-data-paths
         path: /tmp/data_paths.txt
         
 - name: data-converter
   inputs:
     artifacts:
       - name: data_paths
         path: /tmp/raw_data/data_paths.txt
         
   container:
      image: docker/data-checker:latest
      command: [data_converter, /tmp/raw_data/data_paths.txt]

 - name: save-into-db
     .. .. ..

 - name: notify-data-science-team
     .. .. ..

在Argo Workflows中，最基本的概念是工作流（workflow）和模板（template）。工作流对象表示工作流的单个实例，包含工作流的定义和执行状态。我们应该将工作流视为一个“活动”的对象。模板可以被视为函数，它们定义要执行的指令。entrypoint字段定义了主函数，也就是将首先执行的模板。

在代码清单9.2中，我们看到一个包含四个步骤的顺序工作流：check-new-data -> transform_data -> save-into-db -> notify-data-science-team。每个步骤可以引用一个模板，并通过artifacts（文件）传递参数。例如，check-new-data引用了data-checker模板，该模板定义了用于检查是否有新数据的Docker镜像。data-checker模板还声明了步骤输出——新到达的数据文件路径将保存到/tmp/data_paths.txt作为输出值。

接下来，步骤transform_data将check-new-data的输出绑定到data-converter模板的输入。这就是变量在步骤和模板之间传递的方式。一旦您提交了工作流（例如，argo submit -n argo sample_workflow.yaml），您可以使用Argo Workflows的UI或以下命令来查看工作流运行的详细信息：

# list all the workflows
argo list -n argo

# get details of a workflow run
argo get -n argo {workflow_name}

除了使用argo命令，我们还可以使用Kubernetes CLI命令来检查工作流的执行情况，因为Argo Workflows是在Kubernetes上原生运行的；以下是一个示例：

# list all argo customer resource definitions
kubectl get crd -n argo

# list all workflows
kubectl get workflows -n argo

# check specific workflow
kubectl describe workflow/{workflow_name} -n argo

要了解更多关于Argo Workflows的信息，您可以查看Argo Workflows用户指南（mng.bz/WAG0）和Argo Workflows架构图（argoproj.github.io/argo-workfl…）。

代码Docker化：便捷的生产部署 Argo Workflows本质上是一个Kubernetes Pod（Docker镜像）调度系统。尽管它要求将代码编写为一系列的Docker镜像，但它在编排系统内部提供了很大的灵活性和隔离性。由于代码以Docker形式存在，它可以在任何工作节点上执行，而无需担心配置工作节点环境。

Argo Workflows的另一个优势是其低成本的生产部署。当您在Docker中本地测试代码时，Docker镜像（原型代码）可以直接在Argo Workflows中使用。与Airflow不同，Argo Workflows几乎不需要将原型代码转换为生产工作流的工作。

关键特点

Argo Workflows具有以下关键特点：

• 低成本的安装和维护 – Argo Workflows原生运行在Kubernetes上，因此您可以直接使用Kubernetes进程来解决任何问题，无需学习其他工具。此外，它的安装非常简单；通过几个kubectl命令，您就可以在Kubernetes环境中运行Argo Workflows。
• 强大的工作流执行 – Kubernetes Pod为Argo Workflows的任务执行提供了良好的隔离性。Argo Workflows还支持定时工作流和任务重试。
• 模板和组合性 – Argo Workflows的模板类似于函数。在构建工作流时，Argo Workflows支持组合不同的模板（步骤函数）。这种组合性鼓励团队之间共享通用工作，从而极大地提高了生产力。
• 完善的用户界面 – Argo Workflows提供便捷的用户界面，用于管理整个工作流的生命周期，例如提交/停止工作流、列出所有工作流和查看工作流定义。
• 高度灵活和适用 – Argo Workflows定义了用于管理系统和添加新功能（插件）的REST API，并将工作流任务定义为Docker镜像。这些特性使得Argo Workflows具有高度的可定制性，并在许多领域广泛应用，如机器学习、ETL、批处理/数据处理和持续集成/持续交付/持续部署（CI/CD）。
• 适用于生产环境 – Argo Workflows专为严肃的生产环境而设计。Kubeflow Pipeline和Argo CD是生产化Argo Workflows的绝佳示例。

局限性

使用Argo Workflows的深度学习系统的缺点如下：

• 每个人都需要编写和维护YAML文件 – Argo Workflows要求将工作流定义为一个Kubernetes CRD（自定义资源定义）的YAML文件。对于单个项目而言，短小的YAML文件还是可以管理的，但是一旦工作流数量增多并且工作流逻辑变得更加复杂，YAML文件可能会变得冗长和令人困惑。Argo Workflows提供了模板来简化工作流定义，但除非您习惯于使用Kubernetes的YAML配置，否则仍然不太直观。
• 必须成为Kubernetes的专家 – 如果您是Kubernetes的专家，那么使用Argo Workflows可能会感觉自然而然。但是对于新手用户来说，可能需要花费相当多的时间学习Kubernetes的概念和实践。
• 任务执行延迟 – 在Argo Workflows中，对于每个新任务，Argo都会启动一个新的Kubernetes Pod来执行它。Pod的启动可能会导致每个任务的执行时间增加数秒或数分钟，这限制了Argo在支持对时间敏感的工作流时的表现。例如，Argo Workflows不适合具有毫秒级服务级别协议（SLA）的实时模型预测工作流，因为该工作流会处理毫秒级的模型预测请求。

Metaflow

Metaflow是一个注重MLOps的用户友好的Python库。它最初由Netflix开发，并于2019年开源。Metaflow的特点在于它遵循以人为中心的设计原则，不仅用于自动化工作流程，还旨在减少在深度学习项目开发中消耗的人力时间（操作成本）。

在9.1.3节中，我们指出了从原型代码到生产工作流的转换在机器学习开发中引发了很多摩擦。数据科学家必须为每个模型开发迭代构建和测试一个新版本的工作流程。为了弥合原型和生产之间的差距，Metaflow进行了两个改进：首先，简化了工作流程的构建；其次，统一了本地环境和生产环境中的工作流程执行体验（见图9.7）。

在图9.7中，我们可以看到Metaflow将原型和生产环境都视为一流的执行环境。因为Metaflow库提供了一组统一的API来抽象实际的基础架构，无论在哪个环境中运行，工作流程都可以以相同的方式执行。例如，工作流可以通过本地调度程序和生产调度程序运行，而无需进行任何更改。本地调度程序在本地执行工作流程，而生产调度程序则集成到其他生产编排系统中，如AWS Step Functions或Argo Workflows。

Metaflow允许用户使用Python注释来对Python代码进行标注，以定义工作流程。Metaflow库会根据Python注释自动创建/打包工作流程。有了Metaflow的Python注释，Vena可以在不改变她的原型代码的情况下构建工作流程。 除了无缝的工作流程创建和测试，Metaflow还提供了其他有用的功能，对于模型的可重复性非常重要，例如工作流程/步骤的版本控制和步骤的输入/输出保存。如果想了解更多关于Metaflow的信息，可以查阅Metaflow的官方网站（docs.metaflow.org/）以及Ville Tuulos所著的Metaflow书籍《Effective Data Science Infrastructure》（Manning出版社，2022年；www.manning.com/books/effec…）。

典型用户案例

以下是使用Metaflow来自动化之前在9.3.1和9.3.2节中看到的相同数据处理工作的伪代码示例：

# define workflow DAG in a python class
class DataProcessWorkflow(FlowSpec):
  # define "data source" as an input parameter for the workflow
  data_source = Parameter(
      "datasource_path", help="the data storage location for data process"
     , required=True
  )

  @step
  def start(self):
    # The parameter “self.data_source” are available in all steps. 
    self.newdata_path = dataUtil.fetch_new_data(self.data_source)
    self.next(self.transform_data)

  @step
  def transform_data(self):
      self.new_data = dataUtil.convert(self.newdata_path)
      # fan out to two parallel branches after data transfer.   
      self.next(self.save_to_db, self.notify_data_science_team)

  @step
  def save_to_db(self):
      dataUtil.store_data(self.new_data)
      self.next(self.join)

  @step
  def notify_data_science_team(self):
     slackUtil.send_notification(messageUtil.build_message(self.new_data))
     self.next(self.join)

  # join the two parallel branches steps:
  # notify_data_science_team and save_to_db
  
  @step
  def join(self, inputs):
    self.next(self.end)

  @step
  def end(self, inputs):
    # end the flow.
    pass

if __name__ == "__main__":
  DataProcessWorkflow()

在代码清单9.3中，我们可以看到Metaflow采用了一种新颖的方法来构建工作流，使用代码注解来标记函数，并使用self.next函数来连接步骤，我们可以轻松地从我们的原型代码构建一个工作流DAG（图9.8）。

这里的一个优点是我们不需要在单独的系统中定义工作流DAG，并将代码重新打包成不同的格式，比如Docker镜像。Metaflow工作流嵌入在我们的代码中。工作流开发和原型代码开发发生在同一个地方，并且可以从整个机器学习开发周期的开始到结束一起进行测试。 一旦代码准备好了，我们可以在本地验证和运行工作流。请参考以下示例命令：

# display workflow DAG
python data_process_workflow.py show

# run the workflow locally
python data_process_workflow.py run

一旦我们完成了本地开发和测试，就可以通过以下两个命令将工作流推送到生产环境：

# push the workflow from local to AWS step functions
python data_process_workflow.py --with retry step-functions create

# push the workflow from local to Argo workflows
python data_process_workflow.py --with retry argo-workflows create

这些命令将我们在代码清单9.3中定义的数据处理工作流导出到AWS Step Functions和Argo Workflows。然后，您可以在AWS Step Functions UI或Argo Workflows UI中通过名称搜索该工作流，并查看导出的工作流。

注意： Metaflow在本地和生产环境之间提供统一的开发体验。由于Metaflow提供了统一的API，我们在本地和生产环境中测试代码和工作流时具有无缝的体验。无论使用的是Metaflow本地调度程序、Argo Workflows还是AWS Step Functions作为后端工作流编排系统，Metaflow在工作流开发方面的用户体验保持一致！

主要功能

Metaflow提供以下主要功能：

• 将代码结构化为工作流程—Metaflow允许用户通过对Python代码进行注释来创建工作流程，大大简化了工作流程的构建。
• 可复现性—Metaflow保留了执行每个工作流程步骤所需的数据、代码和外部依赖项的不可变快照。Metaflow还记录了每个工作流程执行的元数据。
• 版本控制—Metaflow通过对工作流程中的所有代码和数据进行哈希处理来解决ML项目的版本控制需求。
• 强大的工作流程执行—元数据提供了工作流程级别和步骤级别的依赖管理机制，使用@conda装饰器实现。它还提供了任务重试功能。
• 针对机器学习的易用性设计—Metaflow将原型和生产视为同等重要。它提供了一组统一的API来抽象基础设施，因此相同的代码可以在原型环境和生产环境中运行而无需进行任何更改。
• 无缝可扩展性—Metaflow集成了Kubernetes和AWS Batch，使用户能够轻松定义所需的计算资源，并可以在任意数量的实例上并行执行工作流程步骤。例如，通过将类似@batch(cpu=1, memory=500)的注释应用于步骤函数，Metaflow将与AWS Batch合作分配所需的资源来计算此步骤。

局限性

在深度学习系统中使用Metaflow的缺点如下：

• 不支持条件分支—Metaflow的步骤注释不支持条件分支（只有当条件满足时才执行某个步骤）。这不是一个严重问题，但这是一个很好的功能。
• 没有作业调度程序—Metaflow本身没有作业调度程序，因此无法使用cron工作流。这不是一个大问题，因为Metaflow可以与其他支持作业调度的编排系统集成，如AWS Step Functions和Argo Workflows。
• 与AWS紧密耦合—Metaflow的最重要功能与AWS紧密耦合，例如Amazon S3和AWS Batch。幸运的是，Metaflow是一个开源项目，因此可以将其扩展到非AWS的替代方案中。

何时使用

如果你正在寻找一个用于非机器学习项目自动化工作流执行的编排系统，Airflow和Argo Workflows都是很好的选择。它们都有出色的社区支持，并在IT行业广泛使用。如果你的系统运行在Kubernetes上，并且团队对使用Docker感到舒适，那么Argo Workflows是一个很好的选择；否则，Airflow也不会让你失望。

如果你正在寻找一个用于简化机器学习项目开发的系统，Metaflow是强烈推荐的选择。Metaflow不仅是一个编排工具，更是一个专注于节省数据科学家在机器学习开发周期中时间的MLOps工具。因为Metaflow抽象了机器学习项目的后端基础设施部分，数据科学家可以专注于模型开发，而不必担心生产环境的转换和部署。

总结

• 工作流程是某个更大任务的一系列操作。工作流程可以被视为步骤的有向无环图（DAG）。步骤是最小的可恢复计算单元，描述了要执行的操作；一个步骤要么成功完成，要么完全失败。DAG指定了步骤之间的依赖关系和执行顺序。

• 工作流编排是根据工作流程DAG中定义的顺序执行工作流程步骤。

• 采用工作流程鼓励工作共享、团队协作和自动化。

• 在应用工作流程于深度学习项目时，主要挑战是降低工作流程构建成本，简化工作流程的测试和调试。

• 构建/评估工作流程编排系统的六个推荐设计原则是：重要性、易用性、可扩展性、任务隔离性、可扩展性和以人为中心性。

• 选择非机器学习项目的编排系统时，Airflow和Argo Workflows都是很好的选择。如果项目运行在Kubernetes和Docker上，Argo Workflows是更好的选择。

• 选择机器学习项目的编排系统时，Metaflow目前是最佳选择。