Nephele: Efficient Parallel Data Processing in the Cloud

Proceedings of the 2nd Workshop on Many-Task Computing on Grids and Supercomputers, November 2009

https://dl.acm.org/doi/10.1145/1646468.1646476

在云的时代，对于一个并发数据处理框架的调度器来说，要回答的问题从 Given a set of compute resources, how to distribute the particular tasks of a job among them? 转变为了 Given a job, what compute resources match the tasks the job consists of best?

这种新的范式对调度器提出了三个要求：

  1. 必须了解 job 执行所在的云环境，如各种 VM 的类型及其定价
  2. 必须能够描述一个 job 的各个 task 之间的依赖关系，进而判断何时一个 VM 不再需要可以提前释放
  3. 必须能够决定一个 task 应该在哪种类型的 VM 上执行

这种灵活性同时带来了一些挑战，其中最主要的问题在于网络拓扑，数据处理框架很难知道网络的层级，比如两个节点之间的数据传输经过了几个交换机，对于调度器来说，拓扑感知的调度非常困难。即使知道了底层的网络层级，由于 VM 的迁移特性，网络拓扑可能会随时变动。解决这种问题的唯一方法是将需要大量数据传输的 tasks 调度在一个更强的 VM 上以保证 Data locality。

Nephele

基于云计算带来的机遇和挑战，论文提出了 Nephele，一种基于云环境的全新数据处理框架。其架构遵循 master-worker 模式:

nephele architecture

在一个 job 提交之前，用户需要启动一个实例运行 Job Manager(JM)，它负责接收用户提交的 Job 及调度，它能够通过云提供的服务接口动态申请及释放 VM。

Nephele 框架的另一种角色为 Task Manager(TM)，它从 JM 接收任务，执行之后将完成状态或可能的错误信息汇报给 JM。

一个任务在 Nephele 中通过 DAG 来描述，原因在于:

  1. DAG 允许 task 有多个入度和出度，可以极大简化传统数据结合算子，如 join
  2. DAG 的边表示了执行中的 Job 的数据传输路径，Nephele 可以根据这些信息判断哪些实例可以关闭释放

一个 Nephele Job 的定义由三个手动步骤构成:

  1. the user must write the program code for each task of his processing job or select it from an external library
  2. the task program must be assigned to a vertex
  3. the vertices must be connected by edges to define the communication paths of the job

用户通过 Job Graph 在一个抽象的层级描述任务和任务之间的关系，将任务并行度及任务在实例上的调度留给 Nephele。用户还可以在任务描述中增加注解给 Job Graph 提供更多信息，如:

  • Number of subtasks
  • NUmber of subtasks per instance
  • Sharing instances between tasks
  • Channel types
  • Instance type
nephele architecture

在接收到用户的 Job Graph 后，JM 将其转换为 Execution Graph，这是 Nephele 调度和监控的主要数据结构。相对 Job Graph 只在抽象层次描述任务，Exectution Graph 多了物理层次任务调度到具体实例的信息及 tasks 之间的通信通道:

nephele architecture

其中几个重要的概念:

  • Execution Stage 一个 Exection Stage 执行之前，必须保证其前序 Execution Stage 都已执行完毕，它的另外三个特性使得它可以类比为 checkpoint
    • when the processing of a stage begins, all instances required within the stage are allocated
    • all subtasks included in this stage are set up and ready to receive records
    • before the processing of a new stage, all intermediate results of its preceding stages are stored in a persistent manner
  • Group Vertex 对应一个 Job 的一个 task，如果有并发，则一个 Group Vertex 可以有多个 Execution Vertex
  • Execution Instance 执行实例，多个 Group Vertex 可以调度在同一个 Exectuion Instance 进而提高 Data Locality
  • Channels Nephele 要求所有的边都替换成 channel，包括如下几种:
    • Network channels - 要求通信子任务存在于同一个 Stage
    • In-Memory channels - 要求通信子任务存在于同一个 Stage
    • File channels - Nephele 只允许不同 Stage 的子任务使用这种方式

读后感

Nephele 是 Flink 的前身，这种根据任务申请资源、按量付费的方式在云时代有着巨大的机会，这也是为何 Snowflake、PingCap 这样的厂商逐渐撅起的一个因素。