TensorFlow遇上Spark

TensorFlowOnSpark 项目是由Yahoo开源的一个软件包,实现TensorFlow集群服务部署在Spark平台之上。

大家好,这次我将分享TensorFlow On Spark的解决方案,将TensorFlow集群部署在Spark平台之上,实现了TensorFlow与Spark的无缝连接,更好地解决了两者数据传递的问题。

这次分享的主要内容包括TensorFlowOnSpark架构设计,探讨其工作原理,通过理解其设计,更好地理解TensorFlow集群在Spark平台上的运行机制。

首先,探讨TensorFlowOnSpark的架构与设计。主要包括如下两个基本内容:

  • 架构分析
  • 生命周期

在开始之前,先探讨一下TensorFlowOnSpark的背景,及其它需要解决的问题。为了实现Spark利用TensorFlow深度学习,及其GPU加速的能力,最常见的解决方案如上图所示。

搭建TensorFlow集群,并通过利用既有的Spark集群的数据完成模型的训练,最种再将训练好的模型部署在Spark集群上,实现数据的预测。

该方案虽然实现了Spark集群的深度学习,及其GPU加速的能力,但需要Spark集群与TensorFlow集群之间的数据传递,造成冗余的系统复杂度。

很容易想到,可以将TensorFlow集群部署在Spark之上,用于解决集群间数据传递的问题。

依次类同,该方案可实现Caffe部署在Spark集群之上,实现Spark集群对多种深度学习框架的支持能力,并兼容既有Spark组件的完整性,包括Spark MLLib, Spark Streaming, Spark SQL等。

TensorFlowOnSpark的架构较为简单,Spark Driver程序并不会参与TensorFlow内部相关的计算和处理。其设计思路像是将一个TensorFlow集群运行在了Spark上,其在每个Spark Executor中启动TensorFlow应用程序,然后通过gRPC或RDMA方式进行数据传递与交互。

TensorFlowOnSpark的Spark应用程序包括4个基本过程。

  • Reserve:组建TensorFlow集群,并在每个Executor进程上预留监听端口,启动“数据/控制”消息的监听程序。
  • Start:在每个Executor进程上启动TensorFlow应用程序;
  • Train/Inference:在TensorFlow集群上完成模型的训练或推理
  • Shutdown:关闭Executor进程上的TensorFlow应用程序,释放相应的系统资源(消息队列)。

用户直接通过spark-submit的方式提交Spark应用程序(mnist_spark.py)。其中通过--py_files选项附带TensorFlowOnSpark框架(tfspark.zip),及其TensorFlow应用程序(mnist_dist.py),从而实现TensorFlow集群在Spark平台上的部署。

tfos.part9.9_9.pdf.jpg

首先看看TensorFlow集群的建立过程。首先根据spark-submit传递的num_executor参数,通过调用cluster = sc.parallelize(num_executor)建立一个ParllelCollectionRDD,其中分区数为num_executor。也就是说,此时分区数等于Executor数。

然后再调用cluster.mapPartitions(TFParkNode.reserve)将ParllelCollectionRDD变换(transformation)为MapPartitionsRDD,在每个分区上回调TRSparkNode.reserve。

TRSparkNode.reserve将会在该节点上预留一个端口,并驻留一个Manager服务。Manager持有一个队列,用于完成进程间的同步,实现该节点的“数据/控制”消息的服务。

数据消息启动了两个队列:Input与Output,分别用于RDD与Executor进程之间的数据交换。

控制消息启动了一个队列:Control,用于Driver进程控制PS任务的生命周期,当模型训练完成之后,通过Driver发送Stop的控制消息结束PS任务。

tfos.part10.10_10.pdf.jpg

这是从分区的角度看待TensorFlow集群建立的过程,横轴表示RDD。这里存在两个RDD,第一个为ParllelCollectionRDD,然后变换为MapPartitionsRDD。

纵轴表示同一个分区(Partition),并在每个分区上启动一个Executor进程 。在Spark中,分区数等于最终在TaskScheduler上调度的Task数目。

此处,sc.parallelize(num_executor)生成一个分区数为num_executor的ParllelCollectionRDD。也就是说,此时分区数等于num_executor数目。

在本例中,num_executor为3,包括1个PS任务,2个Worker任务。

tfos.part11.11_11.pdf.jpg

TensorFlow集群建立后,将生成上图所示的领域模型。其中,一个TFCluster将持有num_executor个TFSparkNode节点;在每个TFSparkNode上驻留一个Manager服务,并预留一个监听端口,用于监听“数据/控制”消息。

实际上,TFSparkNode节点承载于Spark Executor进程之上。

tfos.part12.12_12.pdf.jpg

TensorFlow集群建立后,通过调用cluster.start启动集群服务。其结果将在每个Executor进程上启动TensorFlow应用程序。

此处,需要对原生的TensorFlow应用程序进行适配修改,包括2个部分:

  • Feeding与Fetching: 数据输入/输出机制修改
  • ClusterSpec: TF集群的构造描述

其余代码都将保留,最小化TensorFlow应用程序的修改。

tfos.part13.13_13.pdf.jpg

在cluster上调用foreachPartition(TFSparkNode.start(map_func)),将在每个分区(Executor进程)上回调TFSparkNode.start(map_func)。其中,map_func是对应TF应用程序的包装。

通过上述过程,将在Spark上拉起了一个TF的集群服务。从而使得Spark集群拥有了深度学习和GPU加速的能力。

tfos.part14.14_14.pdf.jpg

当Spark平台上已经拉起了TF集群服务之后,便可以启动模型的训练或推理过程了。在训练或推理过程中,最重要的是解决数据的Feeding和Fetching问题。

TFoS上提供了两种方案:

  • TensorFlow QueueRunner:利用TensorFlow提供的FileReader和QueueRunner机制。Spark未参与任何工作,请查阅TensorFlow官方相关文档。
  • Spark Feeding:首先从RDD读取分区数据(通过HadoopRDD.compute),然后将其放在Input队列中,Executor进程再从该队列中取出,并进一步通过feed_dict,调用session.run将分区数据供给给TensorFlow Graph中。
tfos.part15.15_15.pdf.jpg

Feeding过程,就是通过Input Queue同步实现的。当RDD读取分区数据后,阻塞式地将分区数据put到Input队列中;TFGraph在session.run获取Next Batch时,也是阻塞式地等待数据的到来。

tfos.part16.16_16.pdf.jpg

同样的道理,Fetching过程与Feeding过程类同,只是使用Output Queue,并且数据流方向相反。

session.run返回的数据,通过put阻塞式地放入Output Queue,RDD也是阻塞式地等待数据到来。

tfos.part17.17_17.pdf.jpg

以模型训练过程为例,讲解RDD的变换过程。此处以Mnist手写识别为例,左边表示X,右边表示Y。分别通过HadoopRDD读取分区数据,然后通过MapPartititionRDD变换分区的数据格式;然后通过zip算子,实现两个RDD的折叠,生成ZipPartitionsRDD。

然后,根据Epochs超级参数的配置,将该RDD重复执行Epochs次,最终将结果汇总,生成UnionRDD。

在此之前,都是Transformation的过程,最终调用foreachPartition(train)启动Action,触发Spark Job的提交和任务的运行。

tfos.part18.18_18.pdf.jpg

当模型训练或推理完成之后,分别在Input/Control队列中投掷Stop(以传递None实现)消息,当Manager收到Stop消息后,停止队列的运行。

最终,Spark应用程序退出,Executor进程退出,整个工作流执行结束。

tfos.part19.19_19.pdf.jpg
tfos.part20.20_20.pdf.jpg

推荐资料,强烈推荐直接地源代码阅读。

tfos.part21.21_21.pdf.jpg
tfos.part22.22_22.pdf.jpg

原文发布于微信公众号 - 人工智能LeadAI(atleadai)

原文发表时间:2017-09-30

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏一名叫大蕉的程序员

Spark你一定学得会(四)No.68

小伙伴在写Spark任务的过程中感觉非常巨痛苦,总是有奇奇怪怪的问题,写好的程序在开发环境跑得好好的,一上到生产直接被干懵逼了。今天呢,我就跟大伙好好聊聊 Sp...

1735
来自专栏进击的程序猿

Dynamo:Amazon的高可用性的键-值存储系统

Dynamo是一个分布式键值系统,最初用于支持购物车系统,强调的是提供一个“永远在线“的用户体验。

612
来自专栏Golang语言社区

Golang语言社区--【数据库知识】从关系型数据库到非关系型数据库

1. 关系型数据库 关系型数据库,是指采用了关系模型来组织数据的数据库。 关系模型是在1970年由IBM的研究员E.F.Codd博士首先提出的,在之后的几十年中...

3588
来自专栏蘑菇先生的技术笔记

浅谈分布式计算的开发与实现(二)

接上篇,离线计算是对已经入库的数据进行计算,在查询时对批量数据进行检索、磁盘读取展示。 而实时计算是在数据产生时就对其进行计算,然后实时展示结果,一般是秒级。 ...

502
来自专栏Albert陈凯

使用Storm处理事务型实时计算需求时的几处难点

比流量或者订单淘宝可以把我们甩出几条大街。淘宝的兄弟可以自豪地说他们的实时应用已经承受住了双十一全世界范围内最大的单日数据流的冲击。而阿里巴巴中文站的流量和订单...

3417
来自专栏CSDN技术头条

Spark SQL在100TB上的自适应执行实践

Spark SQL是Apache Spark最广泛使用的一个组件,它提供了非常友好的接口来分布式处理结构化数据,在很多应用领域都有成功的生产实践,但是在超大规模...

2986
来自专栏CSDN技术头条

LinkedIn开源大数据计算引擎 Cubert,并为此创建新的语言

【编者按】Linkedin周二宣布开源其大数据计算引擎Cubert,其名字来源于鲁比克方块(Rubik’s Cube),为了让开发人员更容易使用Cubert,而...

1885
来自专栏腾讯大数据的专栏

TDW千台Spark千亿节点对相似度计算

相似度计算在信息检索、数据挖掘等领域有着广泛的应用,是目前推荐引擎中的重要组成部分。随着互联网用户数目和内容的爆炸性增长,对大规模数据进行相似度计算的需求变得...

25910
来自专栏加米谷大数据

大数据基础之Spark

Spark 是 2010 年由 UC Berkeley AMPLab 开源的一款 基于内存的分布式计算框架,2013 年被Apache 基金会接管,是当前大数据...

622
来自专栏祝威廉

如何基于Yarn开发你的分布式程序

这篇文章不会具体教你如何使用Yarn的API,但是会教你我实践过后的一些经验。接下来的内容会探讨以下两个主题:

624

扫描关注云+社区