Apache Spark源码走读之8 -- Spark on Yarn

欢迎转载,转载请注明出处,徽沪一郎。

概要

Hadoop2中的Yarn是一个分布式计算资源的管理平台,由于其有极好的模型抽象,非常有可能成为分布式计算资源管理的事实标准。其主要职责将是分布式计算集群的管理,集群中计算资源的管理与分配。

Yarn为应用程序开发提供了比较好的实现标准,Spark支持Yarn部署,本文将就Spark如何实现在Yarn平台上的部署作比较详尽的分析。

Spark Standalone部署模式回顾

上图是Spark Standalone Cluster中计算模块的简要示意,从中可以看出整个Cluster主要由四种不同的JVM组成

  1. Master 负责管理整个Cluster,Driver Application和Worker都需要注册到Master

  2. Worker 负责某一个node上计算资源的管理,如启动相应的Executor

  3. Executor RDD中每一个Stage的具体执行是在Executor上完成

  4. Driver Application driver中的schedulerbackend会因为部署模式的不同而不同

换个角度来说,Master对资源的管理是在进程级别,而SchedulerBackend则是在线程的级别。

启动时序图

YARN的基本架构和工作流程

YARN的基本架构如上图所示,由三大功能模块组成,分别是1) RM (ResourceManager)
2) NM (Node Manager) 3) AM(Application Master)

作业提交


  1. 用户通过Client向ResourceManager提交Application,
    ResourceManager根据用户请求分配合适的Container,然后在指定的NodeManager上运行Container以启动ApplicationMaster

  2. ApplicationMaster启动完成后,向ResourceManager注册自己

  3. 对于用户的Task,ApplicationMaster需要首先跟ResourceManager进行协商以获取运行用户Task所需要的Container,在获取成功后,ApplicationMaster将任务发送给指定的NodeManager

  4. NodeManager启动相应的Container,并运行用户Task

实例

上述说了一大堆,说白了在编写YARN
Application时,主要是实现ClientApplicatonMaster。实例请参考github上的simple-yarn-app.

Spark on Yarn

结合Spark Standalone的部署模式和YARN编程模型的要求,做了一张表来显示Spark
Standalone和Spark on Yarn的对比。



























Standalone YARN Notes 
Client Client standalone请参考spark.deploy目录
Master ApplicationMaster  
Worker ExecutorRunnable  
Scheduler YarnClusterScheduler  
SchedulerBackend YarnClusterSchedulerBackend  

作上述表格的目的就是要搞清楚为什么需要做这些更改,与之前Standalone模式间的对应关系是什么。代码走读时,分析的重点是ApplicationMaster,
YarnClusterSchedulerBackend和YarnClusterScheduler

一般来说,在Client中会显示的指定启动ApplicationMaster的类名,如下面的代码所示

    ContainerLaunchContext amContainer =
Records.newRecord(ContainerLaunchContext.class);
amContainer.setCommands(
Collections.singletonList(
"$JAVA_HOME/bin/java" +
" -Xmx256M" +
" com.hortonworks.simpleyarnapp.ApplicationMaster" +
" " + command +
" " + String.valueOf(n) +
" 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" +
" 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr"
)
);

但在yarn.Client中并没有直接指定ApplicationMaster的类名,是通过ClientArguments进行了封装,真正指定启动类的名称的地方在ClientArguments中。构造函数中指定了amClass的默认值是org.apache.spark.deploy.yarn.ApplicationMaster

实例说明

将SparkPi部署到Yarn上,下述是具体指令。

$ SPARK_JAR=./assembly/target/scala-2.10/spark-assembly-0.9.1-hadoop2.0.5-alpha.jar     ./bin/spark-class org.apache.spark.deploy.yarn.Client       --jar examples/target/scala-2.10/spark-examples-assembly-0.9.1.jar       --class org.apache.spark.examples.SparkPi       --args yarn-standalone       --num-workers 3       --master-memory 4g       --worker-memory 2g       --worker-cores 1

从输出的日志可以看出,
Client在提交的时候,AM指定的是org.apache.spark.deploy.yarn.ApplicationMaster

13/12/29 23:33:25 INFO Client: Command for starting the Spark ApplicationMaster: $JAVA_HOME/bin/java -server -Xmx4096m -Djava.io.tmpdir=$PWD/tmp org.apache.spark.deploy.yarn.ApplicationMaster --class org.apache.spark.examples.SparkPi --jar examples/target/scala-2.9.3/spark-examples-assembly-0.8.1-incubating.jar --args  ‘yarn-standalone‘  --worker-memory 2048 --worker-cores 1 --num-workers 3 1> /stdout 2> /stderr

小结

spark在提交时,所做的资源申请是一次性完成的,也就是说对某一个具体的Application,它所需要的Executor个数是一开始就是计算好,整个Cluster如果此时能够满足需求则提交,否则进行等待。而且如果有新的结点加入整个cluster,已经运行着的程序并不能使用这些新的资源。缺少rebalance的机制,这点上storm倒是有。

参考资料

  1. Launch Spark On YARN http://spark.apache.org/docs/0.9.1/running-on-yarn.html

  2. Getting started Writing YARN Application http://hortonworks.com/blog/getting-started-writing-yarn-applications/

  3. 《Hadoop技术内幕 深入解析YARN架构设计与实现原理》 董西成著

  4. YARN应用开发流程  http://my.oschina.net/u/1434348/blog/193374 强烈推荐!!!

时间: 2024-12-19 22:46:19

Apache Spark源码走读之8 -- Spark on Yarn的相关文章

Apache Spark源码走读之9 -- Spark源码编译

欢迎转载,转载请注明出处,徽沪一郎. 概要 本来源码编译没有什么可说的,对于java项目来说,只要会点maven或ant的简单命令,依葫芦画瓢,一下子就ok了.但到了Spark上面,事情似乎不这么简单,按照spark officical document上的来做,总会出现这样或那样的编译错误,让人懊恼不已. 今天闲来无事,又重试了一把,居然o了,做个记录,以备后用. 准备 我的编译机器上安装的Linux是archlinux,并安装后如下软件 scala 2.11 maven git 下载源码 第

Apache Spark源码走读之16 -- spark repl实现详解

欢迎转载,转载请注明出处,徽沪一郎. 概要 之所以对spark shell的内部实现产生兴趣全部缘于好奇代码的编译加载过程,scala是需要编译才能执行的语言,但提供的scala repl可以实现代码的实时交互式执行,这是为什么呢? 既然scala已经提供了repl,为什么spark还要自己单独搞一套spark repl,这其中的缘由到底何在? 显然,这些都是问题,要解开这些谜团,只有再次开启一段源码分析之旅了. 全局视图 上图显示了java源文件从编译到加载执行的全局视图,整个过程中最主要的步

Apache Spark源码走读之22 -- Spark MLLib中拟牛顿法L-BFGS的源码实现

欢迎转载,转载请注明出处,徽沪一郎. 概要 本文就拟牛顿法L-BFGS的由来做一个简要的回顾,然后就其在spark mllib中的实现进行源码走读. 拟牛顿法 数学原理 代码实现 L-BFGS算法中使用到的正则化方法是SquaredL2Updater. 算法实现上使用到了由scalanlp的成员项目breeze库中的BreezeLBFGS函数,mllib中自定义了BreezeLBFGS所需要的DiffFunctions. runLBFGS函数的源码实现如下 def runLBFGS( data:

Apache Spark源码走读之1 -- Spark论文阅读笔记

转自:http://www.cnblogs.com/hseagle/p/3664933.html 楔子 源码阅读是一件非常容易的事,也是一件非常难的事.容易的是代码就在那里,一打开就可以看到.难的是要通过代码明白作者当初为什么要这样设计,设计之初要解决的主要问题是什么. 在对Spark的源码进行具体的走读之前,如果想要快速对Spark的有一个整体性的认识,阅读Matei Zaharia做的Spark论文是一个非常不错的选择. 在阅读该论文的基础之上,再结合Spark作者在2012 Develop

Apache Spark源码走读之12 -- Hive on Spark运行环境搭建

欢迎转载,转载请注明出处,徽沪一郎. 楔子 Hive是基于Hadoop的开源数据仓库工具,提供了类似于SQL的HiveQL语言,使得上层的数据分析人员不用知道太多MapReduce的知识就能对存储于Hdfs中的海量数据进行分析.由于这一特性而收到广泛的欢迎. Hive的整体框架中有一个重要的模块是执行模块,这一部分是用Hadoop中MapReduce计算框架来实现,因而在处理速度上不是非常令人满意.由于Spark出色的处理速度,有人已经成功将HiveQL的执行利用Spark来运行,这就是已经非常

Apache Spark源码走读之13 -- hiveql on spark实现详解

欢迎转载,转载请注明出处,徽沪一郎 概要 在新近发布的spark 1.0中新加了sql的模块,更为引人注意的是对hive中的hiveql也提供了良好的支持,作为一个源码分析控,了解一下spark是如何完成对hql的支持是一件非常有趣的事情. Hive简介 Hive的由来 以下部分摘自Hadoop definite guide中的Hive一章 "Hive由Facebook出品,其设计之初目的是让精通SQL技能的分析师能够对Facebook存放在HDFS上的大规模数据集进行分析和查询. Hive大大

Apache Spark源码走读之21 -- 浅谈mllib中线性回归的算法实现

欢迎转载,转载请注明出处,徽沪一郎. 概要 本文简要描述线性回归算法在Spark MLLib中的具体实现,涉及线性回归算法本身及线性回归并行处理的理论基础,然后对代码实现部分进行走读. 线性回归模型 机器学习算法是的主要目的是找到最能够对数据做出合理解释的模型,这个模型是假设函数,一步步的推导基本遵循这样的思路 假设函数 为了找到最好的假设函数,需要找到合理的评估标准,一般来说使用损失函数来做为评估标准 根据损失函数推出目标函数 现在问题转换成为如何找到目标函数的最优解,也就是目标函数的最优化

Apache Spark源码走读之14 -- Graphx实现剖析

欢迎转载,转载请注明出处,徽沪一郎. 概要 图的并行化处理一直是一个非常热门的话题,这里头的重点有两个,一是如何将图的算法并行化,二是找到一个合适的并行化处理框架.Spark作为一个非常优秀的并行处理框架,将一些并行化的算法移到其上面就成了一个很自然的事情. Graphx是一些图的常用算法在Spark上的并行化实现,同时提供了丰富的API接口.本文就Graphx的代码架构及pagerank在graphx中的具体实现做一个初步的学习. Google为什么赢得了搜索引擎大战 当Google还在起步的

Apache Spark源码走读之5 -- DStream处理的容错性分析

欢迎转载,转载请注明出处,徽沪一郎,谢谢. 在流数据的处理过程中,为了保证处理结果的可信度(不能多算,也不能漏算),需要做到对所有的输入数据有且仅有一次处理.在Spark Streaming的处理机制中,不能多算,比较容易理解.那么它又是如何作到即使数据处理结点被重启,在重启之后这些数据也会被再次处理呢? 环境搭建 为了有一个感性的认识,先运行一下简单的Spark Streaming示例.首先确认已经安装了openbsd-netcat. 运行netcatnc -lk 9999 运行spark-s