Spark源码编译并在YARN上运行WordCount实例

193次阅读

共计 4828 个字符，预计需要花费 13 分钟才能阅读完成。

在学习一门新语言时，想必我们都是”Hello World”程序开始，类似地，分布式计算框架的一个典型实例就是 WordCount 程序，接触过 Hadoop 的人肯定都知道用 MapReduce 实现 WordCount，当前内存分布式计算框架 Spark 因为其计算速度之快，并且可以部署到 Hadoop YARN 中运行，已经受到各大公司的青睐，Spark 社区提供了一些编译好的 jar 包，但是其中没有适配 Hadoop-2.2.0 的 jar 包，为了避免版本问题，需要自己编译指定 hadoop 版本的 Spark jar 包。下面介绍如何编译 Spark 源码并在 YARN 上运行 WordCount 程序。

Spark 是由 Scala 编写的，虽然我们也可以通过 Spark 的 Java 或 Python 接口来编写应用程序，但是为了能更好的学习 Spark，建议采用 Scala 来编写应用程序，Scala 是一种函数式编程语言，其简洁和优雅的编程风格相信在不久后会让你喜欢上它的。之前我已经安装配置好 Hadoop 环境，请参考本站前面的一篇文章”CentOS 下 Hadoop-2.2.0 集群安装配置“，下面我们来安装 Scala 编译环境。

Scala 安装配置

去 Scala 官网下载 Scala，我下载的是 Scala-2.10.4（因为官方提供的 spark-1.1.0 编译好的 jar 包都是基于 scala-2.10.4 编译的），将下载后的软件包解压到用户根目录，并配置好环境变量。

[hadoop@master ~]$ tar zxvf scala-2.10.4.tgz
[hadoop@master ~]$ vim ~/.bash_profile
 #添加 Scala 环境变量
 export SCALA_HOME=$HOME/scala-2.10.4
 export PATH=$SCALA_HOME/bin:$PATH

Spark 安装配置

去 Spark 社区下载 Spark 源码，我下载的是 spark-1.1.0 版本，解压，并进入源码根目录，执行以下命令编译：

[hadoop@master ~]$ tar zxvf spark-1.1.0.tgz
[hadoop@master ~]$ cd spark-1.1.0
[hadoop@master ~]$ sbt/sbt assembly -Pyarn -Phadoop-2.2 -Pspark-ganglia-lgpl -Pkinesis-asl -Phive

值得说明的是，我指定编译规则是集成 hadoop2.2 的，以便与我的 Hadoop 环境适配，这里需要等待漫长的编译过程…中途要下载大量的依赖包。编译完成后，可以在目录 assembly 中找到编译打包好的 jar 包：

[hadoop@master scala-2.10]$ ls
spark-assembly-1.1.0-hadoop2.2.0.jar
[hadoop@master scala-2.10]$ pwd
/home/hadoop/spark-1.1.0/assembly/target/scala-2.10

下面我们配置 Spark 的环境变量：

[hadoop@master ~]$ vim ~/.bash_profile
 #添加 Spark 环境变量
 export SPARK_HOME=$HOME/spark-1.1.0
 export PATH=$SPARK_HOME/bin:$PATH

如何让 Spark 能知道我们的 YARN 在哪呢，这里需要在 Spark 配置文件中指定一下 YARN 的位置：

[hadoop@master ~]$ cd $SPARK_HOME/conf/
[hadoop@master conf]$ cp spark-env.sh.template spark-env.sh
[hadoop@master conf]$ vim spark-env.sh
 #添加如下两行，指定你的 java 和 yarn 的位置
 export JAVA_HOME=/home/hadoop/jdk1.7.0_40/
 export YARN_CONF_DIR=/home/hadoop/hadoop-2.2.0/etc/hadoop/

环境都准备好了，下面我们介绍如何构建 Spark WordCount 项目，并将其部署到 YARN 上运行。

构建 WordCount 项目

这里采用 sbt 的方式来构建项目，需要满足 sbt 的项目目录结构（其中 target 目录是编译后自动生成的）：

|--build.sbt
|--lib
|--project
|--src
|   |--main
|   |    |--scala
|   |--test
|         |--scala
|--sbt
|--target

先创建出项目的目录结构，并从 spark 目录中拷贝 sbt 构建工具和前面编译好的 jar 包：

[hadoop@master ~]$ mkdir -p spark-wordcount/lib
[hadoop@master ~]$ mkdir -p spark-wordcount/project
[hadoop@master ~]$ mkdir -p spark-wordcount/src/main/scala
[hadoop@master ~]$ mkdir -p spark-wordcount/src/test/scala
[hadoop@master ~]$ cp -R $SPARK_HOME/sbt spark-wordcount/
[hadoop@master ~]$ cp -R $SPARK_HOME/assembly/target/scala-2.10/spark-assembly-1.1.0-hadoop2.2.0.jar spark-wordcount/lib/

创建 sbt 项目构建文件 build.sbt，该文件指定项目名、scala 版本以及声明依赖，但是本实例先不加入依赖，直接将依赖的 jar 包放到项目的 lib 目录下:

[hadoop@master ~]$ cd spark-wordcount
[hadoop@master spark-wordcount]$ vim build.sbt
name := "WordCount"
[空行]
version := "1.0.0"
[空行]
scalaVersion := "2.10.4"

在 project 目录下新建文件 build.properties，里面指定 sbt 的版本：

[hadoop@master project]$ vim build.properties
sbt.version=0.13.5

到这里，WordCount 项目的准备工作都已做好了，下面我们用 scala 来编写 Spark WordCount 程序，在目录 src/main/scala/ 目录下新建文件 WordCount.scala：

import org.apache.spark._
import SparkContext._

object WordCount {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println("Usage: WordCount <inputfile> <outputfile>");
      System.exit(1);
    }
    val conf = new SparkConf().setAppName("WordCount")
    val sc = new SparkContext(conf)
    val result = sc.textFile(args(0))
                   .flatMap(line => line.split(""))
                   .map(word => (word, 1))
                   .reduceByKey(_ + _)
    result.saveAsTextFile(args(1))
  }
}

代码编写完毕后，开始编译打包，在项目根目录下用 sbt 编译打包：

[hadoop@master spark-wordcount]$ sbt/sbt clean compile package

编译完成后，我们能看到会多出一个 target 目录，里面就包括了编译后的.class 文件和打好的 jar 包：

[hadoop@master spark-wordcount]$ sbt/sbt clean compile package
[hadoop@master spark-wordcount]$ ls target/scala-2.10/
classes  wordcount_2.10-1.0.0.jar

在 YARN 上运行 WordCount

在 hdfs 上首先上传一些待处理的文本：

[hadoop@master spark-wordcount]$ $HADOOP_HOME/bin/hdfs dfs -put ~/test.txt input

编写如下启动脚本，将 Spark WordCount 提交到 YARN 上运行：

#!/usr/bin/env bash

PROJECT_HOME=/home/hadoop/project/spark-wordcount
# 以防万一，确保删除输出目录
$HADOOP_HOME/bin/hdfs dfs -rmr /user/hadoop/output

$SPARK_HOME/bin/spark-submit \
--class WordCount \
--master yarn-cluster \
--num-executors 2 \
--driver-memory 1g \
--executor-memory 1g \
--executor-cores 1 \
$PROJECT_HOME/target/scala-2.10/wordcount_2.10-1.0.0.jar \
hdfs://master:9000/user/hadoop/input \
hdfs://master:9000/user/hadoop/output

这里采用的是 yarn-cluster 部署模式，num-executors 指定启动的 executor 数量，driver-memory 指定 drive 端所需内存大小，executor-memory 指定 executor 所需内存大小，executor-cores 指定每个 executor 使用的内核数。运行如上脚本后，打开 Hadoop 的 web 页面可以看到运行正常，如下图所示：

Spark 源码编译并在 YARN 上运行 WordCount 实例