基于 Hadoop YARN 的 TensorFlow GPU 集群的调度扩展

陆忠华，孙琨,2，王彦棡，王珏，刘芳*

1. 中国科学院计算机网络信息中心，北京 100190

2. 中国科学院大学，北京 100049

引言

近年来，深度学习已经广泛地应用于语音识别，机器翻译，计算机视觉等领域，都取得了很好的效果。随着深度学习理论的兴起，多种深度学习的编程框架不断涌现。TensorFlow、Caffe、PyTorch 等深度学习框架在架构、性能、模型构建等许多方面都差异甚大。每种框架都有其特别适用的领域，而统一管理调度这些框架的平台却非常缺乏，极少数支持的平台，例如腾讯的 DI-X 平台，非开源，对自身的业务支持性很好，但是可扩展性不足。

另一方面，深度学习理论的发展离不开 GPU 硬件的进步。GPU (Graphics Processing Unit，图形处理器)是一种专门的图像运算处理器，主要工作于个人计算机、高性能服务器或移动设备之上，为它们加速各种 3D 图像和特效的处理[1]。对于很多科学计算而言，性能主要取决于 GPU 的浮点计算能力[2]，特别是对深度学习任务来说。但是，由于涉及 GPU 的计算很多本身就需要巨大的计算量，单机通常无法在短时间内完成，因此 GPU 的集群化管理和使用成为一种必然的趋势。

深度学习的模型训练需要海量数据，海量数据的预处理需要用大数据技术进行处理。大数据处理框架中最为流行的当属 Hadoop 和 Spark[3]。Hadoop和 Spark 均支持 Hadoop YARN (Yet Another Resource Negotiator) 作为资源管理器[4]。Hadoop YARN 的出现使得资源的管理和调度更加简单，大大简化了集群结构，并且使其具备了多种任务类型的可扩展性。

在各种强烈的实际需求推动下，无论是支持GPU 计算的深度学习框架 TensorFlow，还是分布式计算平台 Hadoop，都获得了很好的发展。

TensorFlow 是从谷歌诞生的技术框架。通过TensorFlow，开发人员可以不用从底层开始编写代码，而可以通过各种接口，方便快速地实现各种神经网络架构。在工业界和学术界，TensorFlow 均得到了广泛使用。

同样地，在大数据领域，开发人员在 Hadoop 框架的支持下能够实现运行于大规模集群上的分布式应用程序。Hadoop 的核心之一是资源管理器 YARN，YARN 作为通用平台，可以进行多资源管理和调度，使得在其上构建多个多种应用程序成为可能。

随着大数据和深度学习的关系越来越紧密，分开的大数据和深度学习框架已难以无缝满足现实的需要。因此针对大数据和深度学习相结合的研究，包括研究 GPU 资源调度方式、集群可用性、集群稳定性以及集群的可扩展性等，是很有必要和时代性的。以 Hadoop 系统为基础，把 GPU 资源融合至YARN 管理技术中，可便捷地构建出异构环境下的高可用性集群。

在本文之前，已经有许多研究机构和公司在传统的调度系统上与深度学习相结合。比如曙光的SothisAI，IBM 的 LSF，商汤的内部系统，都是从传统的 slurm 调度系统出发，Hadoop 社区的下一代 3.0版本将支持GPU资源的管理和调度。

本文将基于以上几点，结合传统的大数据集群框架 Hadoop YARN，对在 Hadoop 平台上实现 GPU资源的管理与调度并使其具备执行深度学习框架TensorFlow 的能力进行可行性研究及实现。

1 基于 YARN 的 GPU 资源调度

深度学习需要管理和调度的首要资源是 GPU，现有的 Hadoop Yarn 只支持 CPU 和内存的管理和调度。为了使 Yarn 与 TensorFlow 相结合，需要做的首先是扩展 YARN，使其在原有基础上支持 GPU 的管理和调度。

1.1 YARN 的基本架构

YARN 主要由四部分组成，分别是 ResourceManager、NodeManager、ApplicationMaster 和 Container[5]。Y A R N 从整体上可以看做是主从结构，其中ResourceManager 为主结构，NodeManager为从结构。各个 NodeManager 上资源的统一管理和调度是由ResourceManager 负责的。

YARN 的基本架构如图 1 所示。

从图 1 中可以看出，ResourceManager 是一个全局的资源管理器，Client、NodeManager 均与其进行交互，ResourceManager 负责的是整个系统的资源管理和分配。用户通过 Client 提交的每个应用程序，均包含一个 ApplicationMaster。NodeManager可以看做是每个节点上的资源和任务管理器，负责各个节点的资源使用情况以及各个 Container 的运行状态，并与 ResourceManager 进行交互[5]。NodeManager 同时处理 ApplicationMaster 上附加的各种实际请求，例如 Container 的启动、停止等。Container 则可以看做是 YARN 在多维度资源的封装，Container 的具体封装是根据应用程序的需求动态生成的。

1.2 YARN 的 GPU 资源调度算法 DRF

CPU 和内存是当前 Hadoop YARN 已经支持管理和分配的两种资源类型。在配置文件里设置好节点可被使用的 CPU 核数和内存总量，NodeManager会在启动时向 ResourceManager 注册这些信息。为了使 GPU 可以作为资源进行调度，首先需要在配置选项中增加 GPU 选项以及是否使用 GPU，在ResourceManager 中 ResourceType.java 增加 GPU 的资源类型。

图1 YARN 的基本架构图Fig. 1 The basic architecture of YARN

DRF 算法，全称主资源公平调度 (Dominant Resource Fairness) 算法，是 YARN 进行资源管理和调度的主要算法，可以支持多维资源调度。DRF 算法，在管理和调度 CPU 和内存两种资源的情况下，已经被实践证明非常适合应用于多维资源管理和调度的复杂环境中。因此为了使 YARN 调度器可以支持 GPU 的调度，本文主要是在 DRF 的基础上进行扩展。

DRF 算法中，主资源指的是各所需资源在相应总资源中所占比例最大的资源，DRF 算法的基本设计思想是将多维资源管理调度问题转化为单资源管理调度问题，将所有主资源中最小的主资源进行最大化。其算法伪代码如下：

算法 1 DRF 伪代码

举例说明如下。

假设系统总共有 9 个 CPU、18GB RAM 和 2 块GPU，有两个用户 A 和 B，分别运行了两种任务，两种任务所需的资源量分别为 ＜1CPU，4GB，0GPU＞，＜3CPU，1GB，1GPU＞。A 用户每次需要消耗总 CPU的 1/9，内存的 2/9 和 GPU 的 0，内存占的比例最大，所以 A 的主资源为内存。对于用户 B，每次需要消耗总 CPU 的 3/9，总内存的 1/18，总 GPU 的 1/2。因此 B 的主资源为 GPU。

＜a，b，c＞，其中 a 表示 CPU 的个数，b 表示内存的大小，c 表示 GPU 的块数。

由表 1 可以看出，用户 A 获得的总资源量为＜3，12，0＞，可成功运行 3 个任务，用户 B 获取的资源量为 ＜6，2，2＞，可成功运行 2 个任务。系统剩余的资源量为 ＜0，4，0＞。

2 基于 YARN 的 TensorFlow 应用程序

TensorFlow 是现在主流的深度学习框架之一。TensorFlow 可以很好地满足单用户使用，在多用户情况下，会发生资源使用冲突。因此本文把 Hadoop YARN 作为资源调度层，TensorFlow 作为 YARN 之上的应用程序，YARN 调度后的资源供 TensorFlow所使用，可以满足多用户同时使用，极大提高系统的复用率。以下是关键实现步骤。

2.1 技术架构

如图 2 所示，整个系统包括三种组件：

Client：客户端：负责作业的启动和作业执行状态的获取。

ApplicationMaster (AM)：负责输入数据分片、启动及管理 Container、执行日志保存等；

Container：作业实际执行的地方，取代原生TensorFlow 分布式程序需要在每个实际节点上手动启动 PS (Parameter Server) 和 Worker 进程，变为Container 自动启动，并把进程的运行状态定时向 AM进行汇报，同时还负责应用程序的输出等。

其中 PS (parameter server)：负责保存和更新参数，该角色依托于作业采用的 TensorFlow 深度学习框架，仅在分布式模式下启动。Worker：执行作业训练逻辑，负责作业输出保存。

表1 DRF 资源调度Table 1 DRF resource scheduling

图2 TensorFlow on YARN 架构图Fig. 2 TensorFlow on YARN architecture diagram

2.2 具体实现的关键技术

2.2.1 客户端设计

客户端提交应用程序的关键主要有以下几点。

(1) 初始化

对用户提交应用程序脚本的信息进行识别，并进行相应的初始化，主要包括应用程序中，PS 所需的个数和 CPU 核数，worker 所需的个数，CPU 核数和GPU 个数。每个 ps 和 worker 的参数设置均一样，同时包括应用程序的名称，类型，输入，输出。以及启动应用程序的命令。

(2) 对应用程序的监控

Client 首先通过 getNewApplicationResponse() 和getApplicationId() 获取唯一的 Application ID，然后往数据结构 ApplicationSubmissionContext 中放入启动ApplicationMaster 所需的所有信息。

2.2.2 ApplicationMaster 设计

ApplicationMaster 可以看做是相对独立的第三方，与 ResourceManager 和 NodeManager 两个服务都要进行交互，ApplicationMaster 与 ResourceManager交互，可以获得任务所需的计算资源，与 NodeManager交互，可以实际启动计算任务，并对其进行监控。ApplicationMaster 和 ResourceManager、NodeManager之间的交互均可通过相应的 RPC 函数进行注册、通讯。需要特别注意的是，原生的 TensorFlow 分布式程序所需的节点是在代码里固定的，配置方法如下：

采用 YARN 进行调度后，代码里留下的是从环境中获取具体服务器地址的接口。

因此在 ApplicationMaster 需要把从 YARN 调度获取的节点资源转换为相应 JSON 格式。转换方法主要是通过键值对 HashMap。HashMap 中 key 指的是角色，value 指的是实际使用的节点地址，包括 IP 和端口。

2.2.3 利用原生 TensorFlow 分布式框架

现有的 TensorFlow 分布式框架能够支持不够灵活的分布式计算，需要在程序里指定好具体的节点和端口，并且在每个节点均要手动分发，对于集群不具有管理和调度能力。

除却管理和调度能力外，TensorFlow 是一种很好的深度学习编程框架，因此在实际运行计算任务时，在 Container 级别可以直接复用现有的 TensorFlow框架，从而不改变原有程序运行的结果。这要求TensorFlow 框架在每个节点上均得到同样版本的安装。

本文中 YARN 修改不涉及改 TensorFlow 分布式实际操作，只是把集群的管理交给了 YARN，需要的ps 和 worker 个数由代码脚本设置，YARN 获取相应脚本后，向 ResourceManager 获取相应的可用计算服务器资源，然后把 PS 和 Worker 可获取的具体服务器地址包装成 JSON 格式，传送给 TensorFlow 分布式框架，最后在 Container 层具体执行。

3 实验结果与分析

3.1 系统部署环境

3.1.1 硬件部署环境

10 台曙光 W780-G20 GPU 服务器，每台配置 2颗 Intel Xeon 2650v4 处理器，每个单节点配置 8 块P100 GPU 加速卡，总共有 80 块 NVIDIA Tesla P100 GPU 加速卡。计算存储网络方案采用 In fi niband 高速网络，配置 1 台 108 口 56Gb/s FDR 大端口模块化 IB交换机，系统节点之间以 56Gb/s FDR 线速交换，采用大端口模块。

3.1.2 软件部署环境

Centos7.2，CUDA8.0，cudnn-v5.0，Python2.7，nccl2.0，OpenCV2.4.13，TensorFlow1.1.0，Hadoop2.7.3，java 1.8.0_65 等。

3.2 测试程序介绍

3.2.1 数据集

MNIST 数据集是采集阿拉伯数字 0-9 的手写数字数据，每幅图片均为 0 到 9 中 10 个数字的任意一个，黑白像素。

MNIST 数据集可在http：//yann.lecun.com/exdb/mnist/ 获取，它包含了四个部分：

Training set images：train-images-idx3-ubyte.gz(9.9MB，解压后 47MB，包含 60，000 个样本)

Training set labels：train-labels-idx1-ubyte.gz(29KB，解压后 60KB，包含 60，000 个标签)

Test set images：t10k-images-idx3-ubyte.gz(1.6MB，解压后 7.8MB，包含 10，000 个样本)

Test set labels：t10k-labels-idx1-ubyte.gz (5KB，解压后 10KB，包含 10，000 个标签)

3.2.2 应用程序

应用程序主要分为两部分，一部分为单机TensorFlow 程序，另一部分为分布式 TensorFlow程序。

(1) 单机 TensorFlow 程序

单机 TensorFlow 程序，用于手写数字识别。网络架构为卷积神经网络，两层卷积，两层池化。网络结构用 [a，b，c，d]表示，[a，b]表示卷积核的大小，c 表示上一层的深度，d 表示下一层的深度。第一层卷积为 [5，5，1，32]，第二层卷积为 [5，5，32，64]，池化层皆为 [1，2，2，1]。原生的 TensorFlow 分布式程序和使用 YARN 调度的 TensorFlow 分布式程序，两者代码完全一样。单机 TensorFlow 程序主要参数设置如下：batch_size =64，learning_rate = 0.1，training_epochs = 20。

(2) 分布式 TensorFlow 程序

分布式 TensorFlow 程序，用于手写数字识别。网络架构为全连接神经网络，一层隐藏层。输入节点784 个，中间一层隐藏层 100 个，输出节点 10 个。分别编写了原生的 TensorFlow 分布式程序和使用 YARN调度的 TensorFlow 分布式程序，两者参数设置完全一样，主要代码一样，不同之处在于原生 TensorFlow分布式程序使用的节点是在程序里指定好了，并手动在各个指定节点上分别启动的，使用 YARN 调度的 TensorFlow 分布式程序是由 YARN 自动分配的，程序里并不指定到各个具体节点，在一个节点上即可完成。程序主要参数设置如下：batch_size = 100，learning_rate = 0.0005，training_epochs = 20，其他初始随机变量其他 worker 均与第一个 worker 的参数相同。

原生的 TensorFlow 分布式运行方法以下简称原生 TF，使用 YARN 调度的 TensorFlow 分布式运行方法以下简称 YarnTF。

3.3 实验结果

单机下，原生 TF 和 YarnTF 两者都测试了 GPU块数在 1，2，4，8 下的程序完成时间。每种情况均进行了 10 组实验，共计 100 组实验。每种情况的最后结果去除 10 组实验中的最高、最低值，取剩下 8组的平均值。

分布式下，原生 TF 和 YarnTF 两者都测试了单个 ps，1，2，3，4，8 个 worker 下以及两个 ps，1，2，3，4，8 个 worker 下程序完成时间。每种情况均进行 10 组实验，共计 100 组实验，每组实验的程序完成时间取 worker 中最迟完成的时间。每种情况的最后结果均去除 10 组实验中的最高、最低值，取剩下 8 组的平均值。

3.3.1 单机多卡测试结果

测试时运行同样程序，在单机不同 GPU 块数下，程序运行时间。可得表 2 所示结果。

表2 单机多卡下不同框架运行总时间 (单位：秒)Table 2 Total time for running different frames under a single machine

根据表 3 可得单机多卡不同框架下平均每个 epoch运行时间。

以1块 GPU 为基准，可得不同 GPU 块数下的加速比。

表4 单机多卡下不同框架加速比 (单位：秒)Table 4 Speedup ratio of single-machine multi-card under different frames

根据表 4 可得单机多卡下不同框架加速比。

根据图 3 和图 4，可以看出单机多卡下，YarnTF是非常有效的，与原生 TF 具有非常相似的加速比，YarnTF 所花的纯计算机运行时间比原生 TF 要高，主要是 YarnTF 在资源调度要花一定时间，考虑到手工操作时间，实际总时间 YarnTF 要小于原生 TF。

3.3.2 单个 ps 和多个 worker 测试结果

worker 数可以看做是使用的计算节点数，每个worker 均在不同的节点上，并占满每个节点的 8 块GPU。可得表 5 所示结果。

图3 单机多卡不同框架下平均每个 epoch 运行时间Fig. 3 Average epoch running time of single-machine multicardunder different frames

图4 单机多卡下不同框架加速比Fig. 4 Speedup ratio of single-machine multi-card under different frames

根据表 6 可得单 ps 多 worker 下不同框架平均每个 epoch 运行时间。

由图 5 可以看出，不同 worker 下，每个 epoch实际运行时间 YarnTF 均大于原生 TF，这是因为YarnTF 的资源分配是动态的，只有在调度后才能确定程序运行的具体资源，这部分比原生 TF 多了一些时间，另外应用程序在运行过程中均需要与ApplicationMaster 和客户端进行通信，这部分也比原生 TF 多了一些时间。不过考虑到 YarnTF 的资源为自动分配，并在单节点启动，而非手动启动，原生TF 实际总运行时间 (考虑非计算机实际运行时间) 会大于 YarnTF，而且随着 worker 的增加，非计算机实际运行时间会大大增加。

表5 单 ps 多 worker 下不同框架运行总时间 (单位：秒)Table 5 The total running time of different frames under a single ps multi-worker

表6 单ps多worker下不同框架平均每个epoch运行时间(单位：秒)Table 6 Average epoch running time for different frames under a single ps multi-worker

图5 单 ps 多 worker 下不同框架平均每个 epoch 运行时间Fig. 5 Average epoch running time for different frames under a single ps multi-worker

以 1 个 ps，1 个 worker 为基准，可得不同worker 下得加速比。

根据表 7 可得单 ps 多 worker 下不同框架加速比。

如图 6 所示，YarnTF 和原生 TF 的加速比曲线是类似的，表明 YarnTF 实际运行中增加的损耗并不以增加原生 TF 损耗为代价，YarnTF 具有同样的加速效果。

表7 单ps多worker下不同框架加速比 (单位：秒)Table 7 Speedup ratioof Single ps multi-worker in different frames

图6 单 ps 多 worker 下不同框架加速比Fig.6 Speedup ratioof Single ps multi-worker in different frames

3.3.3 两个 ps 和多个 worker 测试结果

worker 数可以看做是使用的计算节点数，每个worker 均在不同的节点上，并占满每个节点的 8 块GPU。可得 2ps 多 worker 下不同框架平均每个 epoch运行时间。

表8 2ps多worker下不同框架运行总时间 (单位：秒)Table 8 Total running time of different frames under 2ps multi-worker

表 9 和表 6 进行对比，2ps 下多个 worker 每个epoch 运行时间在 worker 数较少时，均大于 1ps 下每个 epoch 运行时间，表明此时多个 ps 和多 worker 损耗的通信时间并不值得。2ps 下 8worker，每个 epoch运行时间比 1ps 下 8worker 下速度更快，表明此时2ps 和多 worker 通信之间的损耗低于多 workerGPU加速的价值。

根据表 9 可得 2ps 多 worker 下不同框架平均每个 epoch 运行时间。

图7 的结果与图 5 类似，YarnTF 的计算机实际运行时间均大于原生 TF。以 2ps，1 个 worker 为基准，可得不同 worker 下得加速比。

根据表 10 可得 2ps 多 worker 下不同框架加速比。

图8 表明 2ps 多 worker 下，YarnTF 同样是非常有效的。

4 总结与下一步工作

从图 4、图 6 和图 8 的加速比来看，无论是单机版，还是分布式版本，YarnTF 都是非常有效的。YarnTF 和原生 TF 具有类似的趋势，表明 YarnTF 加速效果与原生 TF 加速效果相似。从图 3、图 5 和图7 来看，无论是单机还是分布式版本，使用 YARN 调度都会比原生 TensorFlow 程序运行时间多，多余损耗主要用在了 YARN 的资源管理和调度上，但是考虑到原生 TensorFlow 分布式程序需要在程序里指定固定节点，并且需要在每个节点上手动式分发，这会浪费大量的非计算机实际使用时间。多用户使用下，YarnTF的可复用性更强，用户不用知道其他用户用了哪些节点，而原生 TensorFlow 用户间需要提前商量好各节点的使用权。因此使用 YARN 调度 TensorFlow 分布式程序会极大提高平台 GPU 的复用能力。

本文实现了大数据框架与深度学习框架的结合，使得 Tensorflow 可以用 YARN 进行调度，实际测试结果表明用 YARN 调度 TensorFlow 是可行的，并且非常有效的。下一步的工作主要有两部分：(1)尽可能减少用 YARN 调度 TensorFlow 分布式程序中资源管理和调度的损耗。(2) 在 YARN 已经支持TensorFlow 的基础上，使其支持更多深度学习框架，例如 Caffe、MXNet 等。

表9 2ps多worker下不同框架平均每个epoch运行时间(单位：秒)Table 9 Average epoch running time for different frames under 2ps multi-worker

表10 2ps多worker下不同框架加速比 (单位：秒)Table 10 Speedup ratio of 2ps multi-worker for different frames

图7 2ps 多 worker 下不同框架平均每个 epoch 运行时间Fig. 7 Average epoch running time for different frames under 2ps multi-worker

图8 2ps 多 worker 下不同框架加速比Fig. 8 Speedup ratio of 2ps multi-worker for different frames

[1]刘德波. 基于 YARN 的 GPU 集群系统研究 [D]. 中山大学, 2014.

[2]丁艺明, 刘波. 利用 GPU 进行高性能数据并行计算 [J].程序员, 2008(4): 97-99.

[3]孙成刚, 李峥, 唐冬冬等. 基于 GPU 的高性能并行计算应用 [J]. 电子信息对抗技术, 2012, 27(2): 69-73.

[4]怀特, 周敏奇, 等. Hadoop 权威指南：第 2 版 [M]. 清华大学出版社, 2011.

[5]董西成. Hadoop 技术内幕: 深入解析 YARN 架构设计与实现原理 [J]. 中国科技信息, 2014(1): 158-158.

[6]张凯, 秦勃, 刘其成. 基于 GPU-Hadoop 的并行计算框架研究与实现 [J]. 计算机应用研究, 2014, 31(8): 2548-2550.

[7]MBARUSHIMANA Emmanuel. 调度和优化大数据计算框架基于 CPU/GPU 集群 [D]. 北京理工大学, 2015.

[8]裴浩. 基于 GPU 的 Hadoop 平台优化实现 [J]. 福建电脑,2017, 33(3): 41-42.

[9]Abadi M. TensorFlow: learning functions at scale[J]. Acm Sigplan Notices, 2016, 51(9): 1-1.

[10]Jue Wang, Fei Gao, Jose Luis Vazquez-Poletti,Jianjiang Li, Preface of High Performance Computing or Advanced Modeling and Simulation of Materials, Computer Physics Communication, Elsevier Publisher, issue 211,2017 (IF: 3.653).

[11]Jue Wang, Chun Liu, Yuehui Huang, Auto tuning for new energy dispatch problem: A case study, Future Generation Computer Systems, Elsevier Publisher,issue 54, 501-506, 2016.1 (IF: 2.430).