Apache Spark源码走读(三)Spark on Yarn &Spark源码编译 &在YARN上运行SparkPi

  1. 云栖社区>
  2. 博客>
  3. 正文

Apache Spark源码走读(三)Spark on Yarn &Spark源码编译 &在YARN上运行SparkPi

许鹏 2016-09-14 18:34:10 浏览3217
展开阅读全文

<一>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 强烈推荐!!!

<二>Spark源码编译

概要

本来源码编译没有什么可说的,对于java项目来说,只要会点maven或ant的简单命令,依葫芦画瓢,一下子就ok了。但到了Spark上面,事情似乎不这么简单,按照spark officical document上的来做,总会出现这样或那样的编译错误,让人懊恼不已。

今天闲来无事,又重试了一把,居然o了,做个记录,以备后用。

准备

我的编译机器上安装的Linux是archlinux,并安装后如下软件

  1. scala 2.11
  2. maven
  3. git

下载源码

第一步当然是将github上的源码下载下来

git clone https://github.com/apache/spark.git

 源码编译

不是直接用maven也不是直接用sbt,而是使用spark中自带的编译脚本make-distribution.sh

export SCALA_HOME=/usr/share/scala
cd $SPARK_HOME
./make-distribution.sh

如果一切顺利,会在$SPARK_HOME/assembly/target/scala-2.10目录下生成目标文件,比如

assembly/target/scala-2.10/spark-assembly-1.0.0-SNAPSHOT-hadoop1.0.4.jar

使用sbt编译

之前使用sbt编译一直会失败的主要原因就在于有些jar文件因为GFW的原因而访问不了。解决之道当然是添加代理才行。

代理的添加有下面几种方式,具体哪种好用,一一尝试吧,对于最新的spark。使用如下指令即可。

export http_proxy=http://proxy-server:port

方法二,设置JAVA_OPTS

JAVA_OPTS="-Dhttp.proxyServer=proxy-server -Dhttp.proxyPort=portNumber"

运行测试用例

既然能够顺利的编译出jar文件,那么肯定也改动两行代码来试试效果,如果知道自己的发动生效没有呢,运行测试用例是最好的办法了。

假设已经修改了$SPARK_HOME/core下的某些源码,重新编译的话,使用如下指令

export SCALA_HOME=/usr/share/scala
mvn package -DskipTests

假设当前在$SPARK_HOME/core目录下,想要运行一下RandomSamplerSuite这个测试用例集合,使用以下指令即可。

export SPARK_LOCAL_IP=127.0.0.1
export SPARK_MASTER_IP=127.0.0.1
mvn -Dsuites=org.apache.spark.util.random.RandomSamplerSuite test

<三>在YARN上运行SparkPi

概要

“spark已经比较头痛了,还要将其运行在yarn上,yarn是什么,我一点概念都没有哎,再怎么办啊。不要跟我讲什么原理了,能不能直接告诉我怎么将spark在yarn上面跑起来,I'm a dummy, just told me how to do it.” 

如果你和我一样是一个对形而上的东西不是太感兴趣,而只纠结于怎么去做的话,看这份guide保证不会让你失望, :)。

前期准备

本文所有的操作基于arch linux,保证下述软件已经安装

  1. jdk
  2. scala
  3. maven

搭建hadoop

hadoop像它的Logo一样,真得是一个体形无比巨大的大象,如果直接入手去搞这个东东的话,肯定会昏上好长一段时间。个人取巧,从storm弄起,一路走来还算平滑。

hadoop最主要的是hdfs和MapReduce Framework,针对第二代的hadoop即hadoop 2这个Framework变成了非常流行的YARN, 要是没听说过YARN,都不好意思说自己玩过Hadoop了。

不开玩笑了,注意上面一段话中最主要的信息就是hdfs和mapreduce framework,我们接下来的所有配置都是围绕这两个主题来的。

创建用户

添加用户组: hadoop,  添加用户hduser

groupadd hadoop
useradd -b /home -m -g hadoop hduser

下载hadoop运行版

假设当前是以root用户登录,现在要切换成用户hduser

su - hduser
id ##检验一下切换是否成功,如果一切ok,将显示下列内容
uid=1000(hduser) gid=1000(hadoop) groups=1000(hadoop)

下载hadoop 2.4并解压

cd /home/hduser
wget http://mirror.esocc.com/apache/hadoop/common/hadoop-2.4.0/hadoop-2.4.0.tar.gz
tar zvxf hadoop-2.4.0.tar.gz

设置环境变量

export HADOOP_HOME=$HOME/hadoop-2.4.0
export HADOOP_MAPRED_HOME=$HOME/hadoop-2.4.0
export HADOOP_COMMON_HOME=$HOME/hadoop-2.4.0
export HADOOP_HDFS_HOME=$HOME/hadoop-2.4.0
export HADOOP_YARN_HOME=$HOME/hadoop-2.4.0
export HADOOP_CONF_DIR=$HOME/hadoop-2.4.0/etc/hadoop

为了避免每次都要重复设置这些变量,可以将上述语句加入到.bashrc文件中。

创建目录

接下来创建的目录是为hadoop中hdfs相关的namenode即datanode使用

mkdir -p $HOME/yarn_data/hdfs/namenode
mkdir -p $HOME/yarn_data/hdfs/datanode

修改Hadoop配置文件

下列文件需要相应的配置

  1. yarn-site.xml
  2. core-site.xml
  3. hdfs-site.xml
  4. mapred-site.xml

切换到hadoop安装目录

$cd $HADOOP_HOME

修改etc/hadoop/yarn-site.xml, 在<configuration>和</configuration>之间添加如下内容,其它文件添加位置与此一致:

<property>
   <name>yarn.nodemanager.aux-services</name>
   <value>mapreduce_shuffle</value>
</property>
<property>
   <name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name>
   <value>org.apache.hadoop.mapred.ShuffleHandler</value>
</property>

etc/hadoop/core-site.xml

<property>
   <name>fs.default.name</name>
   <value>hdfs://localhost:9000</value> <!--YarnClient会用到该配置项-->
</property>

etc/hadoop/hdfs-site.xml

 <property>
   <name>dfs.replication</name>
   <value>1</value>
 </property>
 <property>
   <name>dfs.namenode.name.dir</name>
   <value>file:/home/hduser/yarn_data/hdfs/namenode</value> <!--节点格式化中被用到-->
 </property>
 <property>
   <name>dfs.datanode.data.dir</name>
   <value>file:/home/hduser/yarn_data/hdfs/datanode</value>
 </property>

etc/hadoop/mapred-site.xml

<property>
      <name>mapreduce.framework.name</name>
      <value>yarn</value>
</property>

格式化namenode

$ bin/hadoop namenode -format

启动hdfs相关进程

启动namenode

$ sbin/hadoop-daemon.sh start namenode

启动datanode

$sbin/hadoop-daemon.sh start datanode

启动mapreduce framework相关进程

启动Resource Manager

sbin/yarn-daemon.sh start resourcemanager

启动Node Manager

sbin/yarn-daemon.sh start nodemanager

启动Job History Server

sbin/mr-jobhistory-daemon.sh start historyserver

验证部署

$jps
18509 Jps
17107 NameNode
17170 DataNode
17252 ResourceManager
17309 NodeManager
17626 JobHistoryServer

运行wordCount

验证一下hadoop搭建成功与否的最好办法就是在上面跑个wordcount试试

$mkdir in
$cat > in/file
This is one line
This is another line

将文件复制到hdfs中

$bin/hdfs dfs -copyFromLocal in /in

运行wordcount

bin/hadoop jar ./share/hadoop/mapreduce/hadoop-mapreduce-examples-2.4.0.jar wordcount /in /out

查看运行结果

bin/hdfs dfs -cat /out/*

先歇一会,配置到这里,已经一头汗了,接下来将spark在yarn上的运行,再坚持一小会

在yarn上运行SparkPi

下载spark

下载spark for hadoop2的版本

运行SparkPi

继续以hduser身份运行,最主要的一点就是设置YARN_CONF_DIR或HADOOP_CONF_DIR环境变量

export YARN_CONF_DIR=$HADOOP_HOME/etc/hadoop
SPARK_JAR=./assembly/target/scala-2.10/spark-assembly_2.10-0.9.1-hadoop2.2.0.jar \
./bin/spark-class org.apache.spark.deploy.yarn.Client \
--jar ./examples/target/scala-2.10/spark-examples_2.10-assembly-0.9.1.jar \
--class org.apache.spark.examples.JavaSparkPi \
--args yarn-standalone \
--num-workers 1 \
--master-memory 512m \
--worker-memory 512m \
--worker-cores 1

检查运行结果

运行结果保存在相关application的stdout目录,使用以下指令可以找到

cd $HADOOP_HOME
find . -name "*stdout"

假设找到的文件为./logs/userlogs/application_1400479924971_0002/container_1400479924971_0002_01_000001/stdout,使用cat可以看到结果

cat ./logs/userlogs/application_1400479924971_0002/container_1400479924971_0002_01_000001/stdout
Pi is roughly 3.14028

网友评论

登录后评论
0/500
评论
许鹏
+ 关注