BMS: 基于Airflow的分布式工作流调度管理平台

在电商平台上，欺诈的攻防战时时刻刻都在悄无声息地上演。要想抓住“坏人”，降低资损，往往需要打出一套复杂的“组合拳”——工作流，一般涉及数据清洗、模型推理、异常分析检测、通知告警等多项任务的调度执行。

图1是eBay Risk团队反合谋欺诈的工作流案例示意图。首先ETL阶段，在大数据平台基于用户、商品、交易等多方面数据，计算出合谋欺诈相关的基础数据指标；Offline Model Eng阶段，基于ETL数据，定时调度离线模型推理，判定交易是否存在合谋欺诈并采取相应措施；Anomaly Detection部分，在nous分析平台，基于ETL数据探测其他欺诈如运费、盗号登录等，并通知业务部门。

图1 Buyer Fraud Tagging ETL工作流

电商风控领域中的工作流调度任务存在如下特点：

已有的平台在工作流的管理和调度上，面临着如下痛点与挑战：

面对上述痛点与挑战，eBay Seller Risk团队在工作流的开发和管理的方案上，进行了长期且深入的探索，打造了一个智能高效的解决方案。

当前业内流行的解决方案包括：商用付费产品如UC4， eBay内的产品如Zeta，开源的产品如Airflow, Spring Batch等。

这里我们重点关注以下几项重要指标：

各平台的功能特点如下表1所示:

表1 工作流功能特性对比

BMS(Batch Management System)的目标是基于Apache Airflow开源框架，立足于eBay各基础设施平台，打造自动化的工作流调度管理解决方案。满足如下目标与特性：

BMS目前已经为eBay的30多个团队提供了服务，涵盖了Risk、Payments和Compliance的30多个业务场景，工作流数量达500多个。

图2 BMS Impacts

运行稳定是BMS的基本功，管理高效则是它的独门绝技。较于其他方案，BMS将任务上线时间从数周缩短到数分钟，跨平台调度延迟时间缩短至不到1天，代码开发工作量几乎降为0。

表2 基于BMS工作流管理的指标提升对比

在正式介绍BMS的设计与实现细节之前，有必要了解其中几个关键概念：

BMS平台由三层架构组成：

• Infrastructure层，负责集成企业内部的各基础设施，让调度器的下达的任务能被派发到具体基础平台并执行；

  
  

• BMS Server层，作为调度中枢，上承用户门户，下接基础设施；

  
  

• BMS Portal层，用户创建、调度、管理和运维工作流的门户。

图3 BMS平台架构

BMS的定位是一个轻量级的调度平台。调度是其强项，而任务的具体执行，则只需交给企业内各优秀基础平台。如大数据平台，模型平台，监控告警平台，数据分析平台等，它们各自都有自己的拿手绝活，因此BMS无需重复造轮子，只需联动各个平台，定时定量有序地将工作流的各个任务分发到企业各个基础平台，并追踪任务执行状态。

图4是Risk BU团队关于合谋欺诈批处理方案的对比图：

• 在BMS之前，要分别在大数据平台定时执行ETL任务；在机器学习平台执行模型推理和在分析平台执行异常检测任务。

  
  

• 而应用BMS之后，实现了所谓跨平台调度：通过一个工作流将所有业务单元编排在一起。用户的工作流只需在BMS这个中军帐内运筹帷幄，即可决胜千里。

图4 基于BMS的跨平台调度示意图

当前，Infrastructure层集成的基础设施平台包括：Spark, Hadoop, Kyuubi, Aihub, Elasticsearch, Nous，Tableau 等。用户可以围绕自身业务，在一个平台上把几乎所有eBay平台的离线任务编排在一起。真正做到一个平台，贯穿全业务流程和全技术链路，避免以往在各平台按照时间约定方式进行单独的松散调度。

这样做的好处是：

• 系统更加轻巧健壮，做到调度资源和执行资源的隔离。避免个别重量级任务的消耗完整个调度集群的资源。

  
  

• 系统层级清晰，将任务调度和任务执行解耦，各平台职能清晰，各展所长。同时，避免维护各种业务执行的环境。例如，大数据平台环境，各机器学习模型环境等等。

BMS Server层包括：Airflow集群、BMS Task库以及BMS 工作流部署插件。

Airflow包括四个核心组件: WebServer提供Web界面，Worker复杂执行任务，Scheduler负责调度，Monitor复杂监控任务。BMS选择了基于Kubernetes的云原生Airflow集群搭建方案，能更好地提高集群的弹性、可靠性、可管理性和安全性。

Kubernetes就像一位优秀的指挥家，可以帮助Airflow集群：

• 协调配合各个组件，保证整个集群的稳定运行；

  
  

• 根据集群的负载情况自动调整资源分配，保证任务的高效执行；

  
  

• 标准化的部署和管理方式简化了集群的维护；

  
  

• 多种安全机制保护集群的安全，确保任务的顺利完成。

Airflow Operator是工作流中任务单元的类型，用于执行某个具体的操作。例如：BashOperator可执行一个Bash命令，SimpleHttpOperator可执行一个http请求，HttpSensor可通过http请求检查某个资源状态。下面的代码示例展示的是一个Airflow原生的HttpSensor实例，用于检查数仓某个表在某一天的数据是否已经更新。Airflow虽然提供了各类丰富的Operator/Sensor ，可以支持用户定义出任务单元实例，但并非开箱即用。原因如下：

1. 对于Operator实例的定义，用户必须写Python 代码，无法通过UI交互进行可视化。

   
   

2. 这些通用的task，必须要和企业的基础设施集成，例如SimpleHttpOperator需要解决用户授权问题，需要集成具体的微服务，才能成为一个具体的具备某个具体功能的task；

BMS继承自Airflow基础Operator，结合eBay的基础设施平台和实际业务，并以可视化的方式封装了BMS Task 。图5展示的是名为DW_Table_Data_Readiness_Sensor的Task定义的两种方式。看看左边的基于Operator方式定义的代码，BMS将它的实现变成了填写一个右边UI表单，工作对象是不是立刻变得赏心悦目了？

图5 BMS Sensor Task定义优化

如果面临的欺诈套路比较深，应对它们你可能需要建立一个类似AMLTM Model，包含42个task的超大工作流。它的定义原本需要编写2000多行代码，想想多么令人崩溃。而如今在BMS上，点点鼠标、填填表单，就可以自由自在的创建和管理工作流了，这又是多么令人振奋。

这样一来，我们的用户尤其是非开发人员，从此可以告别调度框架和Spark、Hadoop、ES、模型环境等复杂背景知识，和繁杂的代码说再见了。这样的突变简直堪比手工作业模式向机械化流水线生产模式的飞跃。

当然，变革的背后，是BMS基于Thymeleaf模板引擎技术设计的Airflow Task渲染器。

当前BMS提供丰富的Task类型，俨然成为用户应对各类业务场景百宝箱，包括的task类型主要有：

有了Task代码自动能生成能力后，用户在Job层定义好工作流调度频率和task的依赖关系(如图7 BMS Job中Job Detail部分所示), 可以得到一个描述的Airflow工作流DAG文件。但是，必须经过部署，这个工作流才能上线运行。

Airflow原生的DAG部署，需要将 DAG 文件放置在指定的 DAG 目录中。在面对企业大规部署需求时，该方式易出错、不便于管理、低效而且不利于版本控制。

为此，BMS 实现了一套完备的基于Restful API 框架的自动化DAG部署插件，实现工作流的一键发布。插件将自动完成流程：

1. BMS的DAG部署Restful Service，将自动将DAG文件上传到指定DAG目录；

   
   

2. 解析DAG并持久化到DB，并将其加载到调度系统中。

图6 BMS Job 部署示意图

BMS Portal层作为门户，支持用户对工作流的可视化。这些功能帮助用户更加高效地管理和运维工作流，提高工作效率和质量。

图7 BMS Job(工作流)和Airflow DAG

图7是用户创建的一个真实工作流实例，BMS Portal提供了可视化界面，来支持task定义和工作流编排。其中上图BMS Job图中展示了Job的关键信息，包括: Job的所有者、所属的组织、访问控制等级、调度类型、调度周期、Task列表、Task的依赖关系等。下图展示的是BMS Job相应的Airflow DAG图。

Task定义

在BMS上，Task的定义已经在BMS Task设计与实现部分已经介绍过，不再赘述。

BMS Portal允许用户对工作流进行全生命周期管理，包括：创建、修改、删除、发布等操作。在日常运维中，用户可以查看工作流的运行历史记录（如图8）、日志详情等，进行失败重试等中常见调度问题。

图8 BMS Job Detail

通过提供审计(Audit)功能，BMS支持用户追踪和审计工作流的变更和操作历史。这在涉及上下游多角色甚至多团队协作的工作流管理场景下尤为重要。

相对于其他监控平台，BMS不仅支持用户监控任务本身的成败，还可以通过MonitorTask对于关键数据指标的监控，从而及时发现和解决问题，提高工作流的稳定性和可靠性。图9是Risk BU团队，基于BMS Monitor task定义的监控指标，进一步创建的kibana dashboard，可以帮助团队，监控与反合谋相关的最新交易量、交易额、交易时间等指标。

图9 基于BMS Monitor Task定义的监控视图

至此，以上架构介绍了BMS 工作流的创建与管理。但是随着业务需求的增长，势必有越来越多的task类型上架到BMS上。

如果每次新增和升级task，都需要人工开发UI，更新后端API来适配新的task参数。对其中潜在的大量开发负担，必须未雨绸缪。对此我们继续攻坚克难，通过基于低代码理念的Task自动化注册与管理工具，实现Task前端UI的自动生成和后端Task Airflow代码的自动生成。

实际上该自动化工具将BMS打造成了一个Task管理的开放平台——类似苹果App Store的BMS Task商店，为开发者提供了task创建的自动化工具。

图10 BMS Task开发上架流程对比

此前，上线新的Task流程如下：

1. 实现新Task的实例化方法的template代码；

2. 为新Task开发相应的UI界面；

3. 更新task 参数传递api和task 模板渲染的上下文，支持新类型task的渲染。

该工具问世后，流程简化为：

• Task元数据注册；

• UI自动生成和渲染上下文更新。

BMS Task Metadata驱动的UI代码自动生成模块，是基于开源框架react-jsonschema-form的表单代码自动生成解决方案（图11）。JSON Schema是一种用于描述JSON数据结构的语言，自动生成表单组件的框架。

图11 react-jsonschema-form

为了提高开发效率，BMS封装了表单元素Metadata编辑器，取代了通过手写JSON Schema、 UISchema来定义表单元素的方式。支持用户通过UI交互创定义UI 表单，也即生成定义一个form所需的前端代码。如图12所示。

图12 基于react-jsonschema-form表单定义代码和BMS 定义表单UI工具

介绍Task前端代码自动生成原理后，下面看一下与此相对应的后端的Task Airflow DAG代码自动生成的核心技术：基于元数据的BMS Task渲染引擎，其工作流程如图13所示：

图13 BMS Task Airflow Python代码自动生成过程

1. 用户在Task UI上填写Task参数值，包括specific attributes列表，然后提交Task 表单，创建Task对象。

   
   

2. 将元数据和Task实例参数统一上传并加载到Task Airflow 渲染引擎的上下文。

   
   

3. Task Airflow 渲染引擎则根据Task Type加载对应的Task Template，将Task对象渲染到Task Template中，从而自动生成Airflow Task实例化代码。如图14所示。

图14 BMS Task 模板到Task Airflow 代码渲染示意图

在本文中，我们讨论了 eBay 如何基于开源框架Airflow，构建了面向批处理工作流的可视化管理平台。深入介绍了eBay BMS团队引入前后端代码的自动生成技术，以低代码方案实现了BMS task 前后端代码生成的自动化。类似于苹果的App Store，这一创新构建了BMS Task商店，为BMS 将来成为公司级或者行业内Task共享平台打下了技术基础。这展示了 eBay 作为技术型公司，通过运用先进技术解决行业内常见的调度疑难问题的见解。

BMS正积极探索推出BMS AI Bot，这是由人工智能驱动的智能批处理工作流设置助手，专为业务运营而设计。该助手通过交互式聊天机器人简化复杂流程，创建个性化、自动化的工作流程。可以根据用户描述从BMS任务库中推荐适当任务并初始化必要的任务信息，引导用户创建工作流程，期望帮助用户更加简单高效地创建调度并管理Workflow。