Uber是如何管理大规模数据工作流的？

让这些数据可操作&＃xff0c;需要进行数据提取、转换、解析和编排&＃xff0c;从而在传统商业智能、机器学习、模型训练、可视化和报表等场景中广泛应用。尽管在Uber迅速发展的初期&＃xff0c;上线了广覆盖面的数据工作流系统&＃xff0c;用户须针对每种用例选择几种工具叠加使用。尽管此大型工具箱可实现敏捷且响应迅速的增长&＃xff0c;但事实证明&＃xff0c;它难以管理和维护&＃xff0c;需要工程师在应对不同项目时&＃xff0c;学习重复数据工作流系统。因此&＃xff0c;Uber需要一个可以创建、管理、调度和部署数据工作流的中央工具。

利用Uber之前部署的各种工具&＃xff0c;包括基于Airflow的平台&＃xff0c;Uber的技术开始开发与Uber规模相称的系统。这项工作使Uber开发了集中式工作流管理系统Piper&＃xff0c;该系统使Uber的数据工作流大众化&＃xff0c;并使从城市运营团队到机器学习工程师的每个人都能更快速、更高效地开展工作。

统一工作流管理系统之路

直到几年前&＃xff0c;Uber的团队使用了多个数据工作流系统&＃xff0c;其中一些基于开源项目&＃xff0c;如Apache Oozie、Apache Airflow和Jenkins&＃xff0c;也有一些用Python和Clojure编写的定制解决方案。每个需要移动数据的用户都必须学习和选择使用不同的系统&＃xff0c;这取决于他们需要完成的特定任务。每个系统都需要根据额外的维护和操作工作来保持其运行、排除问题、修复Bug以及教育用户。

经过一番思考&＃xff0c;Uber决定聚集到单个工作流程系统上。在评估行业中可用的数据工作流工具以及权衡了每种工具的利弊&＃xff0c;并考虑了多个因素&＃xff0c;如易用性、稳定性、开源生态系统、对Hadoop生态系统的依赖性、领域专用语言&＃xff08;DSL&＃xff09;的表达能力以及所使用的编程语言&＃xff08;是否与我们的用户群的语言技能匹配&＃xff09;之后&＃xff0c;Uber研发了新的工作流系统。在新系统中&＃xff0c;Uber寻找以下特点&＃xff1a;

• 工作流应该易于通过代码编写进行创建&＃xff0c;同时也要具有表现力并支持动态生成工作流。

• 支持工程师习惯的开发流程&＃xff0c;包括将数据工作流开发为代码&＃xff0c;并通过版本修订控制进行跟踪。

• 工作流易于可视化、容易管理。

• 工作流日志易于访问&＃xff0c;便于查看工作流的历史及当前运行状态。

经过这次评估&＃xff0c;并以融合支持Uber数据规模的单个工作流程系统为目标&＃xff0c;Uber选择了基于Airflow的系统。基于Airflow的DSL提供了灵活性、表现力和易用性的最佳平衡&＃xff0c;同时可供包括数据科学家、开发人员、机器学习专家和运营人员在内的广泛用户群体使用。

当Uber朝单一数据多租户工作流系统发展时&＃xff0c;开始停用了原有类似系统的工作。弃用这些系统极大地简化了用户工作流的编写&＃xff0c;以及提高团队管理能力和系统长期改进的能力。

选择部署模型&＃xff1a;集中式多租户平台

在部署系统时&＃xff0c;可以选择采用单一的集中管理安装&＃xff0c;而不是为公司每个团队或组织进行安装。Uber找到了两个行业内的案例&＃xff0c;从单一的亚马逊AWS数据管道托管安装到Google Cloud Composer的Airflow分布式安装。在后一种情况下&＃xff0c;用户或管理员负责系统的设置和配置&＃xff0c;而卖方仅提供有限的支持。

Uber考虑了以下几个因素&＃xff0c;才做出此决定&＃xff1a;

• 引入更改后&＃xff0c;系统如何升级&＃xff1f;

• 发生错误或基础结构问题时&＃xff0c;谁负责on-call和系统支持&＃xff1f;

• 随着工作流程数量的增加&＃xff0c;谁来负责扩展系统&＃xff1f;

• 对于不同版本的不同配置系统&＃xff0c;如何避免雪花式的集群和节点&＃xff1f;

• 系统是否需要使用它的团队进行任何维护或配置&＃xff1f;

考虑到上述因素后&＃xff0c;Uber构建了一个集中部署的模型&＃xff0c;即由数据工作流管理团队支持的每个数据中心进行一次单一安装。对于终端用户而言&＃xff0c;Piper&＃xff08;在该系统上的实现&＃xff09;是自助且可靠的&＃xff0c;但是由Uber的技术团队对Piper进行管理&＃xff0c;以确保其能进行更新并能够随着公司工作流程的增长而稳定扩展。还通过功能需求说明&＃xff0c;文档和培训为终端用户提供支持&＃xff0c;他们无需了解系统的内部工作原理。

在选择集中部署模型时&＃xff0c;因为需要确保Piper扩展到支持比每个团队部署时所需的工作流大得多的数量&＃xff0c;因此每个团队部署只需要有限数量的工作流。与多安装模型相比&＃xff0c;Piper还需要提供更好的隔离和多租户功能。

系统架构

虽然Piper基于原始的开源Airflow架构&＃xff0c;但Uber重构了大多数系统&＃xff0c;使其性能更高&＃xff0c;可用性更好&＃xff0c;并适合Uber的基础架构。下图详细说明了Piper的初始结构体系&＃xff1a;

图1&＃xff1a;Piper最初的架构基于Airflow&＃xff08;一个开源的集创建、调度和监控与一体的解决方案&＃xff09;

Piper的原始架构包含以下五个组件&＃xff1a;

• Web服务器&＃xff1a;为HTTP请求提供服务的应用程序服务器&＃xff0c;包括针对UI端点以及JSON API端点的HTTP请求。

• 调度程序&＃xff1a;负责调度工作流程和任务。调度程序考虑了各种因素&＃xff0c;如调度间隔&＃xff0c;任务依赖性&＃xff0c;触发规则和重试次数&＃xff0c;并使用此信息来计算下一组要运行的任务。一旦资源可用&＃xff0c;它将把排队等待的任务在适当的执行器&＃xff08;即示例中的Celery&＃xff09;中执行。

• Celery Worker&＃xff1a;容器执行所有工作流程任务。每个容器从队列&＃xff08;即示例中的Redis&＃xff09;中拉取下一个要执行的任务&＃xff0c;并在本地执行。&＃xff08;可执行任务由工作流ID、任务ID和执行日期进行标识&＃xff09;。

• 元数据数据库&＃xff1a;为系统中所有实体&＃xff08;如工作流、任务、连接、变量和XCOM&＃xff09;记录真实来源&＃xff0c;以及工作流的执行状态。

• Python工作流&＃xff1a;用户编写的Python文件&＃xff0c;用于定义工作流、任务和库。

与用户代码隔离

通过在生产环境操作&＃xff0c;Piper吸取的重要教训是&＃xff0c;需要将用户代码与系统代码隔离。工作流DSL的一个优点是&＃xff0c;工作流可以由任何任意的Python构造定义&＃xff0c;如&＃xff0c;循环访问磁盘上的配置文件&＃xff0c;调出外部服务以获取配置数据或将其在命令行执行。但是&＃xff0c;这种灵活性与系统稳定性和可靠性相冲突&＃xff0c;因为用户代码可以运行任意逻辑&＃xff0c;执行速度慢&＃xff0c;并可能导致系统错误。通常&＃xff0c;良好的系统设计鼓励尽可能将用户代码与系统代码隔离。如下图2所示&＃xff0c;原始体系结构依赖于在所有系统组件&＃xff08;包括调度程序&＃xff0c;Web服务器和Celery worker&＃xff09;中执行用户代码。

图2&＃xff1a;用户代码在所有系统组件中执行&＃xff0c;可能会对Piper的可用性和性能产生负面影响。

Piper的目标之一是尽可能可靠、快速地调度任务。但是&＃xff0c;用户代码在各种组件中运行为系统稳定性带来了诸多问题。在Uber的环境中&＃xff0c;工作流驱动程序可以为单个Python文件生成数千个工作流&＃xff0c;有时会等待外部服务来检索配置&＃xff0c;且加载速度慢&＃xff0c;从而对系统可用性和性能产生负面影响。

基于这些思考&＃xff0c;意识到可以将工作流的元数据表示与Python定义解耦。单个工作流定义文件可以分解为两个单独的表示形式&＃xff1a;

1. 工作流和任务的元数据表示&＃xff1a;序列化表示&＃xff0c;包括工作流/任务属性以及任务之间的依赖关系图。此表示可由诸如调度程序和Web服务器之类的系统组件使用&＃xff0c;它们仅需要了解有关每个工作流程的高级元数据&＃xff0c;而无需加载或执行用户管道定义文件。

2. 完全实例化的工作流&＃xff1a;用户提供的完全实例化的Python任务和工作流表示形式&＃xff0c;符合DSL规范。因此可以使用此表示形式来提取工作流元数据并在Celery worker中执行任务。

图3&＃xff1a;使用Piper&＃xff0c;可以将Pipeline定义视为两种表示形式&＃xff1a;序列化的元数据表示形式和完全可执行的工作流表示形式。

为了实现元数据分解&＃xff0c;在系统中引入了一个新组件&＃xff0c;其作用是加载用户Python工作流程的定义、提取&＃xff0c;然后将其序列化的元数据表示形式存储在数据库中。元数据表示可以由系统的其他组件&＃xff08;如调度程序和Web服务器&＃xff09;使用&＃xff0c;而不必加载任何用户代码。在将工作流的元数据表示与可执行表示分离时&＃xff0c;能够将许多系统组件与必须加载Python工作流定义解耦&＃xff0c;从而使系统更可靠、性能更高。

图4&＃xff1a;不再需要在调度程序或Web服务器中加载用户代码。使用Piper&＃xff0c;工作流序列化程序通过从用户代码中提取元数据来提供隔离。

重构以实现高可用性和水平伸缩弹性

通过元数据序列化与用户代码隔离之后&＃xff0c;Uber希望进一步提高Piper的可伸缩性和系统可用性。Uber的目标是实现&＃xff1a;

• 提高系统效率和语言支持&＃xff1a;在Uber&＃xff0c;对Go和Java用于微服务的使用进行了标准化&＃xff0c;因此&＃xff0c;在Piper中也选择遵循这种语言标准化&＃xff0c;并提供较低的调度延迟和性能改进&＃xff0c;同时仍将在Python中保留DSL。

• 高可用性和消除单点故障&＃xff1a;Uber在Apache Mesos / μDeploy系统上托管服务&＃xff0c;运行在Apache Mesos集群上的Docker容器内。这些服务必须在不宕机的情况下&＃xff0c;妥善处理容器崩溃&＃xff0c;重启和容器重定位。在现有的系统体系结构中&＃xff0c;调度是单点故障&＃xff1a;如果调度程序节点消失&＃xff0c;系统将停止调度任何任务。单点故障也发生在节点重新分配期间&＃xff0c;通常由部署&＃xff0c;硬件维护或资源短缺引起。

• 调度的水平可伸缩性&＃xff1a;现有系统仅支持在任何时间运行单个调度程序。随着新工作流的添加&＃xff0c;调度延迟往往会随着时间的推移而增加。我们希望能够添加额外的调度程序&＃xff0c;自动接管部分运行的工作流。这将提供自动故障转移、减少的调度延迟以及跨多个节点负载均衡作业调度的能力。

为了实现这些目标&＃xff0c;应用了分布式系统概念来提高Piper的可用性和可伸缩性。引入了使用分布式协调服务来提供可用于强化系统的原语&＃xff0c;并通过以下更改对系统进行重构&＃xff1a;

• 用Java重写Piper的调度程序和执行程序组件&＃xff1a;自发布Piper以来&＃xff0c;调度程序和执行程序与加载用户Python代码解耦&＃xff0c;因此现在可以自由使用最适合此job的任何平台或工具。随着Uber在Java和Go上的使用进行标准化&＃xff0c;又用Java重写了调度程序和执行程序组件&＃xff0c;从而能够使用Java性能更高的并发语义来提高系统效率。

• 利用主要运行程序的选举&＃xff1a;对于要作为单例运行的任何系统组件&＃xff0c;如序列化程序和执行程序&＃xff0c;都增加了主要运行程序的选举功能。如果主要运行程序变得不可用&＃xff0c;将从可用的备份节点中自动选出新的主要运行程序。这消除了单点故障&＃xff0c;还减少了Apache Mesos中的部署、节点重启或节点重定向期间的宕机时间。

• 引入工作分区&＃xff1a;回想一下&＃xff0c;目标是能够添加更多调度程序&＃xff0c;并将这些调度程序自动分配给一部分工作流程并对其进行调度。使用分布式协调服务&＃xff0c;能够为任务调度实现有效的工作分区。这种方法可以随时向Piper添加新的调度程序。新的调度程序上线后&＃xff0c;会自动为其分配一组工作流&＃xff0c;并对工作流进行调度。当调度程序节点联机或脱机时&＃xff0c;工作流集将被自动调整&＃xff0c;从而为任务调度提供了高可用性和水平伸缩性。

图5&＃xff1a;高可用性&＃xff0c;解耦和完全分布式的Piper架构设计&＃xff0c;支持多个调度程序同时运行&＃xff0c;同时消除单点故障。

系统开发完成并进入测试阶段后&＃xff0c;决定使用部分迁移策略&＃xff0c;而不是整体迁移所有工作流。首先并排部署了Python和分布式Java调度程序&＃xff0c;并能够在工作流级别切换调度模式。通过这种策略&＃xff0c;能够完全迁移所有工作流程&＃xff0c;而不会影响终端用户。通过上述更改&＃xff0c;现在已经获得了所有系统组件的高可用性和工作流调度的水平可伸缩性&＃xff0c;并通过Java并发提高了性能。

其他平台增强功能

尽管到目前为止的主题涵盖了在调度程序和工作流序列化上执行的主要重构&＃xff0c;但同时Uber的技术团队也将一些其他增强功能集成到了平台&＃xff0c;这些增强功能概述如下&＃xff1a;

• 多种数据中心语义&＃xff1a;目前在每个数据中心运行一次Piper安装。添加了工作流语义&＃xff0c;用户可以指定在单计算模式还是双计算模式下运行工作流。对于数据中心故障&＃xff0c;系统自动将工作流转移到其他数据中心&＃xff0c;而无需用户干预。

• 多租户&＃xff1a;由于有数万个用户和数百个团队在使用该系统&＃xff0c;因此需要使Piper成为多租户使用模式。同时 &＃xff0c;还为工作流、连接和变量之类的许可实体添加了更多语义&＃xff0c;以确保对适当所有者的访问控制。

• 审计&＃xff1a;诸如编辑连接、编辑变量和切换工作流之类的用户操作将保存到审计存储中&＃xff0c;以便在需要时进行搜索。

• 回调&＃xff1a;已实现了通用的回调功能&＃xff0c;该功能使用户只需在UI界面中通过几次点击即可为任何现有工作流程创建和管理回调。

• 视觉创作工具&＃xff1a;在Uber&＃xff0c;有几类用户需要创建工作流&＃xff0c;其中一些用户可能不熟悉Python开发。因此&＃xff0c;提供了几种创建工作流的方法&＃xff0c;通常是通过领域特定的UI动态创建工作流。这些UI创建工具特定于垂直领域&＃xff0c;如机器学习&＃xff0c;仪表板和数据提取。目前还在开发通用的可视化拖放工具&＃xff0c;用于在系统顶层创建工作流程。

• 工作流程定义REST API&＃xff1a;增加了通过REST API调用动态创建工作流程的功能&＃xff0c;而无需使用Python代码。这类似于Apache Storm Flux API。

• 持续部署&＃xff1a;使用monorepo存储工作流程定义代码。确保将monorepo连续部署到必要的系统组件中&＃xff0c;而无需用户干预。

• 持续集成&＃xff1a;每个用户提交到monorepo的工作流&＃xff0c;都需要进行一系列的单元测试&＃xff0c;以确保不引入任何错误并确保工作流的有效性。

• 指标和监控&＃xff1a;已经使用Uber的M3和uMonitor系统插入了指标和监控。还添加了canary工作流&＃xff0c;以评估系统运行状况和性能&＃xff0c;收集有关系统和基础结构统计信息&＃xff0c;并使用这些指标在系统出现故障时发出警报。

• 日志记录&＃xff1a;对任务日志进行了重新设计&＃xff0c;以适应Uber日志记录基础架构&＃xff0c;并确保它对最终用户可靠且即时可用&＃xff0c;而不会影响系统可用性或可靠性。

   

关键总结

从最初的系统部署到今天&＃xff0c;Uber已经从数十个工作流发展为数以万计的工作流&＃xff0c;这些工作流每天由数万个用户管理着成千上万的任务。在保持系统稳定和性能的同时做到了这一点&＃xff0c;并提高了可用性&＃xff0c;可伸缩性和易用性。通过遵循以下原则&＃xff0c;重构造了系统&＃xff0c;以支持Uber不断扩大的规模&＃xff1a;

• 优先考虑用户友好性和编写工作流的表现力。包括轻松管理工作流和即时访问工作流日志。

• 尽可能在整个公司范围内集成到统一的产品。集成到统一的系统中可以大大减少维护负担&＃xff0c;on-call事件以及用户的困惑&＃xff0c;同时可以让团队统一围绕一个产品进行迭代改进。

• 选择中央多租户部署模型。在案例中&＃xff0c;使用的模型使之能更好地支持Uber的用户并提供简便的升级路径&＃xff0c;而无需团队在创建系统的新实例时了解系统内部结构或配置。

• 尽可能避免在系统组件中运行用户代码。分流工作流元数据极大地提高了我们系统的可靠性&＃xff0c;并且还提供了额外的灵活性&＃xff08;允许用Java重写写调度组件&＃xff09;。

• 消除任何单点故障&＃xff0c;确保正常运行时间和系统可用性。

• 使用分布式系统概念&＃xff0c;如主要运行程序选举、故障转移和工作分区&＃xff0c;以提高可用性和可伸缩性。