转载

Meson：支持Netflix的机器学习工作流

Netflix的目标是能预测顾客之所想观看的电影，也即推荐预测。为了做到这一点，每天会运行大量机器学习工作流，而为了支撑创建这么多机器学习工作流和有效利用资源，Netflix的工程师开发了Meson。

Meson是一个通用的工作流和调度框架，它可以跨异质性系统进行ML管道管理、执行工作流。Meson维护构建、训练和验证个性化算法（视频推荐等）的ML管道的生命周期。

Meson的主要目标之一是提高算法实验的速度、稳定性和可重复性，同时允许工程师使用他们自己选择的技术。

机器学习管道的强大面孔

Netflix当前几种生成机器学习管道的关键角色是Spark, MLlib, Python, R和Docker。

下面来了解下一个典型的视频推荐的机器学习管道，以及它们在Meson中是如何表示和处理的。

Meson：支持Netflix的机器学习工作流

工作流涉及到如下几点：

Meson：支持Netflix的机器学习工作流

选择用户集：通过Hive查询来选择用户集来做人群行为分析；
清晰／预处理数据：使用Python脚本创建两个用户集来确保并行路径；
并行路径：一个路径使用Spark构建和分析全局模型，并保存到HDFS作为临时存储。另外一个使用R来构建区域模型。区域的数量是基于人群选择分析而动态变化。在流程图中，Build Regional Model和Validate Regional Model 对于每个区域是重复的，运行时扩展，并且以不同的参数集来执行，见下图；
验证：Scala代码用来测试两个并行路径收敛时的模型稳定性。在这步中，如果模型不稳定，我们也需要再返回、重复整个流程；
发布新模型：建立一个Docker容器来发布新模型，以便被其它生产系统使用。

上图展示了一个运行的工作流（前面描述的）：

用户集选择和数据预处理已完成，图中绿色部分；
流程中的并行路径：
- Spark分支完成模型生成和验证；
- for-each分支分出四个不同的区域模型，它们都在运行中，图中黄色部分。
模型选择的Scala部分已激活，如图中蓝色部分所示。这表明一个或者多个输入分支已完成，但是它仍然没有被调度执行。其原因是输入分支还有未启动或者正在进行的。
运行时环境和参数会根据业务策略传递到工作流当中。

深度解密

接下来让我们揭开Meson神秘的面纱，去理解Meson是如何跨系统运行和生态系统不同组件间是怎样交互的。工作流在运行的过程中有各种资源要求和期望，Meson是利用 Apache Mesos 来实现的。Mesos提供CPU、内存、存储和其它计算资源的任务级别的隔离和抽象。Meson利用这点特性来实现对任务的高扩展性和容错性。

Meson：支持Netflix的机器学习工作流

Meson调度器

Meson调度器被注册为一个Mesos框架，进行管理各种工作流的启动、流控制和运行时。Meson把实际的资源调度发到Mesos，包括各种内存和CPU要求。Meson依赖Mesos进行资源调度，但是Meson的调度器设计成可插入的，所以也可以选择其它框架来做资源调度。

每步一旦被调度，Meson调度器从Mesos申请到实际的资源，并把任务注册到Mesos master。

Meson Executor

Meson Executor是一个定制化的Mesos Executor。写入一个executor允许我们保持Meson间通讯通道，这对长连接任务（框架消息发送到Meson调度器）非常有用。Meson Executor也使得使用者可以传入定制化的丰富数据，而不仅仅传入一些退出代码或者状态信息。

一旦Mesos调度一个Meson任务，它会在slave端下载所有任务依赖后启动一个Meson executor。当核心任务被执行，executor会发送心跳、完成百分比、状态信息等。

DSL

Meson提供一个基于Scala的DSL，使得开发人员更容易的使用和创建定制化的工作流。下面展示前述工作流是如何使用DSL来定义：

val getUsers = Step("Get Users", ...) val wrangleData = Step("Wrangle Data", ...) ... val regionSplit = Step("For Each Region", ...) val regionJoin = Step("End For Each", ...) val regions = Seq("US", "Canada", "UK_Ireland", "LatAm", ...) val wf = start -> getUsers -> wrangleData ==> (    trainGlobalModel -> validateGlobalModel,    regionSplit **(reg = regions) --< (trainRegModel, validateRegModel) >-- regionJoin ) >== selectModel -> validateModel -> end // If verbs are preferred over operators val wf = sequence(start, getUsers, wrangleData) parallel {    sequence(trainGlobalModel, validateGlobalModel)    sequence(regionSplit,              forEach(reg = regions) sequence(trainRegModel, validateRegModel) forEach,              regionJoin) } parallel sequence(selectModel, validateModel, end)

Meson架构扩展

Meson很容易扩展，增加定制化的步骤和扩展功能，比如，Spark Submit Step, Hive Query Step、Netflix的微服务或者其它像Cassandra的系统。

Artifacts

在Meson里，工作流的每步输出都作为“一等公民”对待，并存储为Artifacts。工作流会根据artifact id的存在与否来决定每步的重试是否可跳过。我们也可以通过Meson UI来对Artifacts进行定制的可视化。比如，存储特征重要性作为一个artifact，然后插入一个定制的可视化来比较过去N天特征的重要性。

Meson：支持Netflix的机器学习工作流