Spark集群安装和使用

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

本文主要记录 CDH5 集群中 Spark 集群模式的安装过程配置过程并测试 Spark 的一些基本使用方法。

安装环境如下：

操作系统：CentOS 6.5
Hadoop 版本：cdh-5.4.0
Spark 版本：cdh5-1.3.0_5.4.0

关于 yum 源的配置以及 Hadoop 集群的安装，请参考使用 yum 安装 CDH Hadoop 集群。

1. 安装

首先查看 Spark 相关的包有哪些：

$ yum list |grep spark
spark-core.noarch                  1.3.0+cdh5.4.0+24-1.cdh5.4.0.p0.52.el6
spark-history-server.noarch        1.3.0+cdh5.4.0+24-1.cdh5.4.0.p0.52.el6
spark-master.noarch                1.3.0+cdh5.4.0+24-1.cdh5.4.0.p0.52.el6
spark-Python.noarch                1.3.0+cdh5.4.0+24-1.cdh5.4.0.p0.52.el6
spark-worker.noarch                1.3.0+cdh5.4.0+24-1.cdh5.4.0.p0.52.el6
hue-spark.x86_64                   3.7.0+cdh5.4.0+1145-1.cdh5.4.0.p0.58.el6

以上包作用如下：

spark-core: spark 核心功能
spark-worker: spark-worker 初始化脚本
spark-master: spark-master 初始化脚本
spark-python: spark 的 Python 客户端
hue-spark: spark 和 hue 集成包
spark-history-server

在已经存在的 Hadoop 集群中，选择一个节点来安装 Spark Master，其余节点安装 Spark worker，例如：在 cdh1 上安装 master，在 cdh1、cdh2、cdh3 上安装 worker：

# 在 cdh1 节点上运行
$ sudo yum install spark-core spark-master spark-worker spark-python spark-history-server -y

# 在 cdh1、cdh2、cdh3 上运行
$ sudo yum install spark-core spark-worker spark-python -y

安装成功后，我的集群各节点部署如下：

cdh1 节点:  spark-master、spark-worker、spark-history-server
cdh2 节点:  spark-worker 
cdh3 节点:  spark-worker

2. 配置

2.1 修改配置文件

设置环境变量，在 .bashrc 或者 /etc/profile 中加入下面一行，并使其生效：

export SPARK_HOME=/usr/lib/spark

可以修改配置文件 /etc/spark/conf/spark-env.sh，其内容如下，你可以根据需要做一些修改，例如，修改 master 的主机名称为 cdh1。

# 设置 master 主机名称
export STANDALONE_SPARK_MASTER_HOST=cdh1

设置 shuffle 和 RDD 数据存储路径，该值默认为 /tmp。使用默认值，可能会出现No space left on device 的异常，建议修改为空间较大的分区中的一个目录。

export SPARK_LOCAL_DIRS=/data/spark

如果你和我一样使用的是虚拟机运行 spark，则你可能需要修改 spark 进程使用的 jvm 大小（关于 jvm 大小设置的相关逻辑见 /usr/lib/spark/bin/spark-class）：

export SPARK_DAEMON_MEMORY=256m

更多 spark 相关的配置参数，请参考 Spark Configuration。

2.2 配置 Spark History Server

在运行 Spark 应用程序的时候，driver 会提供一个 webUI 给出应用程序的运行信息，但是该 webUI 随着应用程序的完成而关闭端口，也就是说，Spark 应用程序运行完后，将无法查看应用程序的历史记录。Spark history server 就是为了应对这种情况而产生的，通过配置，Spark 应用程序在运行完应用程序之后，将应用程序的运行信息写入指定目录，而 Spark history server 可以将这些运行信息装载并以 web 的方式供用户浏览。

创建 /etc/spark/conf/spark-defaults.conf：

cp /etc/spark/conf/spark-defaults.conf.template /etc/spark/conf/spark-defaults.conf

添加下面配置：

spark.master=spark://cdh1:7077
spark.eventLog.dir=/user/spark/applicationHistory
spark.eventLog.enabled=true
spark.yarn.historyServer.address=cdh1:18082

如果你是在 hdfs 上运行 Spark，则执行下面命令创建 /user/spark/applicationHistory 目录：

$ sudo -u hdfs hadoop fs -mkdir /user/spark
$ sudo -u hdfs hadoop fs -mkdir /user/spark/applicationHistory
$ sudo -u hdfs hadoop fs -chown -R spark:spark /user/spark
$ sudo -u hdfs hadoop fs -chmod 1777 /user/spark/applicationHistory

设置 spark.history.fs.logDirectory 参数：

export SPARK_HISTORY_OPTS="$SPARK_HISTORY_OPTS -Dspark.history.fs.logDirectory=/tmp/spark -Dspark.history.ui.port=18082"

创建 /tmp/spark 目录：

$ mkdir -p /tmp/spark
$ chown spark:spark /tmp/spark

如果集群配置了 kerberos，则添加下面配置：

HOSTNAME=`hostname -f`
export SPARK_HISTORY_OPTS="$SPARK_HISTORY_OPTS -Dspark.history.kerberos.enabled=true -Dspark.history.kerberos.principal=spark/${HOSTNAME}@LASHOU.COM -Dspark.history.kerberos.keytab=/etc/spark/conf/spark.keytab -Dspark.history.ui.acls.enable=true"

2.3 和 Hive 集成

Spark 和 hive 集成，最好是将 hive 的配置文件链接到 Spark 的配置文件目录：

$ ln -s /etc/hive/conf/hive-site.xml /etc/spark/conf/hive-site.xml

2.4 同步配置文件

修改完 cdh1 节点上的配置文件之后，需要同步到其他节点：

scp -r /etc/spark/conf  cdh2:/etc/spark
scp -r /etc/spark/conf  cdh3:/etc/spark

3. 启动和停止

3.1 使用系统服务管理集群

启动脚本：

# 在 cdh1 节点上运行
$ sudo service spark-master start

# 在 cdh1 节点上运行，如果 hadoop 集群配置了 kerberos，则运行之前需要先获取 spark 用户的凭证
# kinit -k -t /etc/spark/conf/spark.keytab spark/cdh1@JavaCHEN.COM
$ sudo service spark-history-server start

# 在 cdh2、cdh3 节点上运行
$ sudo service spark-worker start

停止脚本：

$ sudo service spark-master stop
$ sudo service spark-worker stop
$ sudo service spark-history-server stop

当然，你还可以设置开机启动：

$ sudo chkconfig spark-master on
$ sudo chkconfig spark-worker on
$ sudo chkconfig spark-history-server on

3.2 使用 Spark 自带脚本管理集群

另外，你也可以使用 Spark 自带的脚本来启动和停止，这些脚本在 /usr/lib/spark/sbin 目录下：

$ ls /usr/lib/spark/sbin
slaves.sh        spark-daemons.sh  start-master.sh  stop-all.sh
spark-config.sh  spark-executor    start-slave.sh   stop-master.sh
spark-daemon.sh  start-all.sh      start-slaves.sh  stop-slaves.sh

在 master 节点修改 /etc/spark/conf/slaves 文件添加 worker 节点的主机名称，并且还需要在 master 和 worker 节点之间配置无密码登陆。

# A Spark Worker will be started on each of the machines listed below.
cdh2
cdh3

然后，你也可以通过下面脚本启动 master 和 worker：

$ cd /usr/lib/spark/sbin
$ ./start-master.sh
$ ./start-slaves.sh

当��，你也可以通过 spark-class 脚本来启动，例如，下面脚本以 standalone 模式启动 worker：

$ ./bin/spark-class org.apache.spark.deploy.worker.Worker spark://cdh1:18080

3.3 访问 web 界面

你可以通过 http://cdh1:18080/ 访问 spark master 的 web 界面。

Spark 集群安装和使用

访问 Spark History Server 页面：http://cdh1:18082/。

Spark 集群安装和使用

注意：我这里使用的是 CDH 版本的 Spark，Spark master UI 的端口为18080，不是 Apache Spark 的 8080 端口。CDH 发行版中 Spark 使用的端口列表如下：

7077 – Default Master RPC port
7078 – Default Worker RPC port
18080 – Default Master web UI port
18081 – Default Worker web UI port
18080 – Default HistoryServer web UI port

4. 测试

Spark 可以以本地模式运行，也支持三种集群管理模式：

Standalone – Spark 原生的资源管理，由 Master 负责资源的分配。
Apache Mesos – 运行在 Mesos 之上，由 Mesos 进行资源调度
Hadoop YARN – 运行在 Yarn 之上，由 Yarn 进行资源调度。

另外 Spark 的 EC2 launch scripts 可以帮助你容易地在 Amazon EC2 上启动 standalone cluster.

在集群不是特别大，并且没有 mapReduce 和 Spark 同时运行的需求的情况下，用 Standalone 模式效率最高。

Spark 可以在应用间（通过集群管理器）和应用中（如果一个 SparkContext 中有多项计算任务）进行资源调度。

4.1 Standalone 模式

该模式中，资源调度是 Spark 框架自己实现的，其节点类型分为 Master 和 Worker 节点，其中 Driver 节点运行在 Master 节点中，并且有常驻内存的 Master 进程守护，Worker 节点上常驻 Worker 守护进程，负责与 Master 通信。

Standalone 模式是 Master-Slaves 架构的集群模式，Master 存在着单点故障问题，目前，Spark 提供了两种解决办法：基于文件系统的故障恢复模式，基于 Zookeeper 的 HA 方式。

Standalone 模式需要在每一个节点部署 Spark 应用，并按照实际情况配置故障恢复模式。

你可以使用交互式命令 spark-shell、pyspark 或者 spark-submit script 连接到集群，下面以 wordcount 程序为例：

$ spark-shell --master spark://cdh1:7077
scala> val file = sc.textFile("hdfs://cdh1:8020/tmp/test.txt")
scala> val counts = file.flatMap(line => line.split("")).map(word => (word, 1)).reduceByKey(_ + _)
scala> counts.count()
scala> counts.saveAsTextFile("hdfs://cdh1:8020/tmp/output")

如果运行成功，可以打开浏览器访问 http://cdh1:4040 查看应用运行情况。

运行过程中，可能会出现下面的异常：

14/10/24 14:51:40 WARN hdfs.BlockReaderLocal: The short-circuit local reads feature cannot be used because libhadoop cannot be loaded.
14/10/24 14:51:40 ERROR lzo.GPLNativeCodeLoader: Could not load native gpl library
java.lang.UnsatisfiedLinkError: no gplcompression in java.library.path
	at java.lang.ClassLoader.loadLibrary(ClassLoader.java:1738)
	at java.lang.Runtime.loadLibrary0(Runtime.java:823)
	at java.lang.System.loadLibrary(System.java:1028)
	at com.hadoop.compression.lzo.GPLNativeCodeLoader.<clinit>(GPLNativeCodeLoader.java:32)
	at com.hadoop.compression.lzo.LzoCodec.<clinit>(LzoCodec.java:71)
	at java.lang.Class.forName0(Native Method)
	at java.lang.Class.forName(Class.java:249)
	at org.apache.hadoop.conf.Configuration.getClassByNameOrNull(Configuration.java:1836)
	at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:1801)
	at org.apache.hadoop.io.compress.CompressionCodecFactory.getCodecClasses(CompressionCodecFactory.java:128)

解决方法可以参考 Spark 连接 Hadoop 读取 HDFS 问题小结这篇文章，执行以下命令，然后重启服务即可：

cp /usr/lib/hadoop/lib/native/libgplcompression.so $JAVA_HOME/jre/lib/amd64/
cp /usr/lib/hadoop/lib/native/libhadoop.so $JAVA_HOME/jre/lib/amd64/
cp /usr/lib/hadoop/lib/native/libsnappy.so $JAVA_HOME/jre/lib/amd64/

使用 spark-submit 以 Standalone 模式运行 SparkPi 程序的命令如下：

$ spark-submit --class org.apache.spark.examples.SparkPi  --master spark://cdh1:7077 /usr/lib/spark/lib/spark-examples-1.3.0-cdh5.4.0-hadoop2.6.0-cdh5.4.0.jar 10

需要说明的是：Standalone mode does not support talking to a kerberized HDFS，如果你以 spark-shell --master spark://cdh1:7077 方式访问安装有 kerberos 的 HDFS 集群上访问数据时，会出现下面异常:

15/04/02 11:58:32 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 6, bj03-bi-pro-hdpnamenn): java.io.IOException: Failed on local exception: java.io.IOException: org.apache.hadoop.security.AccessControlException: Client cannot authenticate via:[TOKEN, KERBEROS]; Host Details : local host is: "cdh1/192.168.56.121"; destination host is: "192.168.56.121":8020;
        org.apache.hadoop.net.NetUtils.wrapException(NetUtils.java:764)
        org.apache.hadoop.ipc.Client.call(Client.java:1415)
        org.apache.hadoop.ipc.Client.call(Client.java:1364)
        org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:206)
        com.sun.proxy.$Proxy17.getBlockLocations(Unknown Source)

4.2 Spark On Mesos 模式

参考 http://dongxicheng.org/framework-on-yarn/apache-spark-comparing-three-deploying-ways/。

4.3 Spark on Yarn 模式

Spark on Yarn 模式同样也支持两种在 Yarn 上启动 Spark 的方式，一种是 cluster 模式，Spark driver 在 Yarn 的 application master 进程中运行，客户端在应用初始化完成之后就会退出；一种是 client 模式，Spark driver 运行在客户端进程中。Spark on Yarn 模式是可以访问配置有 kerberos 的 HDFS 文件的。

CDH Spark 中，以 cluster 模式启动，命令如下：

$ spark-submit --class path.to.your.Class --deploy-mode cluster --master yarn [options] <app jar> [app options]

CDH Spark 中，以 client 模式启动，命令如下：

$ spark-submit --class path.to.your.Class --deploy-mode client --master yarn [options] <app jar> [app options]

以 SparkPi 程序为例：

$ spark-submit --class org.apache.spark.examples.SparkPi \
    --deploy-mode cluster  \
    --master yarn  \
    --num-executors 3 \
    --driver-memory 4g \
    --executor-memory 2g \
    --executor-cores 1 \
    --queue thequeue \
    /usr/lib/spark/lib/spark-examples-1.3.0-cdh5.4.0-hadoop2.6.0-cdh5.4.0.jar \
    10

另外，运行在 YARN 集群之上的时候，可以手动把 spark-assembly 相关的 jar 包拷贝到 hdfs 上去，然后设置 SPARK_JAR 环境变量：

$ hdfs dfs -mkdir -p /user/spark/share/lib
$ hdfs dfs -put $SPARK_HOME/lib/spark-assembly.jar  /user/spark/share/lib/spark-assembly.jar

$ SPARK_JAR=hdfs://<nn>:<port>/user/spark/share/lib/spark-assembly.jar

5. Spark-SQL

Spark 安装包中包括了 Spark-SQL，运行 spark-sql 命令，在 cdh5.2 中会出现下面异常：

$ cd /usr/lib/spark/bin
$ ./spark-sql
java.lang.ClassNotFoundException: org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver
	at java.net.URLClassLoader$1.run(URLClassLoader.java:202)
	at java.security.AccessController.doPrivileged(Native Method)
	at java.net.URLClassLoader.findClass(URLClassLoader.java:190)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:306)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:247)
	at java.lang.Class.forName0(Native Method)
	at java.lang.Class.forName(Class.java:247)
	at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:319)
	at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75)
	at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

Failed to load Spark SQL CLI main class org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.
You need to build Spark with -Phive.

在 cdh5.4 中会出现下面异常：

Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.hive.cli.CliDriver
  at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
  at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
  at java.security.AccessController.doPrivileged(Native Method)
  at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
  at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
  at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
  at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
  ... 18 more

从上可以知道 Spark-SQL 编译时没有集成 Hive，故需要重新编译 spark 源代码。

编译 Spark-SQL

以下内容参考编译 Spark 源代码。

下载 cdh5-1.3.0_5.4.0 分支的代码：

$ git clone git@github.com:cloudera/spark.git
$ cd spark
$ git checkout -b origin/cdh5-1.3.0_5.4.0

使用 maven 编译，先修改根目录下的 pom.xml，添加一行 <module>sql/hive-thriftserver</module>：

<modules>
    <module>core</module>
    <module>bagel</module>
    <module>graphx</module>
    <module>mllib</module>
    <module>tools</module>
    <module>streaming</module>
    <module>sql/catalyst</module>
    <module>sql/core</module>
    <module>sql/hive</module>
    <module>sql/hive-thriftserver</module> <!-- 添加的一行 -->
    <module>repl</module>
    <module>assembly</module>
    <module>external/twitter</module>
    <module>external/kafka</module>
    <module>external/flume</module>
    <module>external/flume-sink</module>
    <module>external/zeromq</module>
    <module>external/mqtt</module>
    <module>examples</module>
  </modules>

然后运行：

$ export MAVEN_OPTS="-Xmx2g -XX:MaxPermSize=512M -XX:ReservedCodeCacheSize=512m"
$ mvn -Pyarn -Dhadoop.version=2.6.0-cdh5.4.0 -Phive -Phive-thriftserver -DskipTests clean package

如果编译成功之后，会在 assembly/target/scala-2.10 目录下生成：spark-assembly-1.3.0-cdh5.4.0.jar，在 examples/target/scala-2.10 目录下生成：spark-examples-1.3.0-cdh5.4.0.jar，然后将 spark-assembly-1.3.0-cdh5.4.0.jar 拷贝到 /usr/lib/spark/lib 目录，然后再来运行 spark-sql。

但是，经测试 cdh5.4.0 版本中的 spark 的 sql/hive-thriftserver 模块存在编译错误，最后无法编译成功，故需要等到 cloudera 官方更新源代码或者等待下一个 cdh 版本集成 spark-sql。

虽然 spark-sql 命令用不了，但是我们可以在 spark-shell 中使用 SQLContext 来运行 sql 语句，限于篇幅，这里不做介绍，你可以参考 http://www.infoobjects.com/spark-sql-schemardd-programmatically-specifying-schema/。