Spark1.0.x入门指南

1 节点说明

HDFS集群和Spark集群之间节点共用。

2 安装HDFS

    见HDFS2.X和Hive的安装部署文档：http://www.cnblogs.com/Scott007/p/3614960.html

3 Spark部署

    Spark常用的安装部署模式有Spark On Yarn和Standalone，可以同时使用。

3.1 Spark on Yarn

    这种模式，借助Yarn资源分配的功能，使用Spark客户端来向Yarn提交任务运行。只需将Spark的部署包放置到Yarn集群的某个节点上即可(或者是Yarn的客户端，能读取到Yarn集群的配置文件即可)。Spark本身的Worker节点、Master节点不需要启动。

    但是，Spark的部署包须是基于对应的Yarn版本正确编译后的，否则会出现Spark和Yarn的兼容性问题。

    on Yarn的两种运行方式，其运行结束后的日志不能在Yarn的Application管理界面看到，目前只能在客户端通过：

yarn logs -applicationId <applicationId>

命令查看每个Application的日志。

3.1.1 配置

    部署这种模式，需要修改conf目录下的spark-env.sh文件。在其中新增如下配置选项：

export HADOOP_HOME= /home/hadoop/hadoop-2.0.0-cdh4.5.0

export HADOOP_CONF_DIR= $HADOOP_HOME/etc/hadoop

SPARK_EXECUTOR_INSTANCES=2

SPARK_EXECUTOR_CORES=1

SPARK_EXECUTOR_MEMORY=400M

SPARK_DRIVER_MEMORY=400M

SPARK_YARN_APP_NAME="Spark 1.0.0"

    其中：

(1) HADOOP_HOME:当前节点中HDFS的部署路径，因为Spark需要和HDFS中的节点在一起；

(2) HADOOP_CONF_DIR:HDFS节点中的conf配置文件路径，正常情况下此目录为$HADOOP_HOME/etc/hadoop；

(3) SPARK_EXECUTOR_INSTANCES:在Yarn集群中启动的Worker的数目，默认为2个；

(4) SPARK_EXECUTOR_CORES:每个Worker所占用的CPU核的数目；

(5) SPARK_EXECUTOR_MEMORY:每个Worker所占用的内存大小；

(6) SPARK_DRIVER_MEMORY:Spark应用程序Application所占的内存大小，这里的Driver对应Yarn中的ApplicationMaster；

(7) SPARK_YARN_APP_NAME：Spark Application在Yarn中的名字；

    配置完成后，将Spark部署文件放置到Yarn的节点中即可。这里，将spark-1.0.0整个目录放到Yarn集群的一个节点192.168.1.112的/home/hadoop(设为spark的安装路径的父目录)路径下。

3.1.2 测试

    在Spark的部署路径的bin路径下，执行spark-submit脚本来运行spark-examples包中的例子。执行如下：

./bin/spark-submit --master yarn 
--class org.apache.spark.examples.JavaWordCount 
--executor-memory 400M 
--driver-memory 400M 
/home/hadoop/spark-1.0.0/examples/target/scala-2.10/spark-examples-1.0.0-hadoop2.0.0-cdh4.5.0.jar ./hdfs-site.xml

    这个例子是计算WordCount的，例子被打包在/home/hadoop/spark-1.0.0/examples/target/scala-2.10/spark-examples-1.0.0-hadoop2.0.0-cdh4.5.0.jar包中，对应的Class为org.apache.spark.examples.JavaWordCount，./hdfs-site.xml是HDFS中指定路径下的一个文件，WordCount就是针对它来做的。而--master yarn就是指定运行在Yarn集群中，以yarn模式运行。

    Spark On Yarn有两种运行模式，一种是Yarn Cluster方式，一种是Yarn Client方式。

(1) Yarn Cluster: Spark Driver程序将作为一个ApplicationMaster在YARN集群中先启动，然后再由ApplicationMaster向RM申请资源启动 executor以运行Task。因为Driver程序在Yarn中运行，所以程序的运行结果不能在客户端显示，所以最好将结果保存在HDFS上，客户端 的终端显示的是作为Yarn的job的运行情况。

(2) Yarn Client: Spark Driver程序在客户端上运行，然后向Yarn申请运行exeutor以运行Task，本地程序负责最后的结果汇总等。客户端的Driver将应用提交 给Yarn后，Yarn会先后启动ApplicationMaster和executor，另外ApplicationMaster和executor都 是装载在container里运行，container默认的内存是1G，ApplicationMaster分配的内存是driver- memory，executor分配的内存是executor-memory。同时，因为Driver在客户端，所以程序的运行结果可以在客户端显 示，Driver以进程名为SparkSubmit的形式存在。

    上面命令中的提交方式“yarn”就是默认按照“Yarn Client”方式运行。用户可自定义运行方式，通过“--master”指定程序以yarn、yarn-cluster或者yarn-client中的一种方式运行。

    需要重点说明的是最后文件的路径，是相当于HDFS中的/user/hadoop而言，hadoop是当前命令的用户。“./hdfs-site.xml”在HDFS中的全路径为“hdfs://namespace/user/hadoop/hdfs-site.xml”，其中hadoop是当前的用户，namespace是HDFS的命名空间；如果写成“/hdfs-site.xml”则在HDFS中指的是“hdfs://namespace/hdfs-site.xml”；当然也可以直接传入“hdfs://namespace/user/hadoop/hdfs-site.xml”用于指定在HDFS中的要进行WordCount计算的文件。

    另外，Spark应用程序需要的CPU Core数目和内存，需要根据当前Yarn的NodeManager的硬件条件相应设置，不能超过NodeManager的硬件条件。

./bin/spark-submit --master yarn 
--class org.apache.spark.examples.JavaWordCount 
--executor-memory 400M 
--driver-memory 400M 
/home/hadoop/spark-1.0.0/examples/target/scala-2.10/spark-examples-1.0.0-hadoop2.0.0-cdh4.5.0.jar hdfs://namespace/user/hadoop/hdfs-site.xml

在Yarn的ResourceManager对应的Web界面中查看启动的Application。

Running:

Success:

同时可以在启动脚本的客户端看到WordCount的运行结果：

3.2 Spark Standalone

    这种模式，就是把Spark单独作为一个集群来进行部署。集群中有两种节点，一种是Master，另一种是Worker节点。Master负责分配任务给Worker节点来执行，并负责最后的结果合并，Worker节点负责具体的任务执行。

3.2.1 配置

    所需修改的配置文件除了spark-env.sh文件以外，还有slave文件，都位于conf目录中。

    slave文件中保存的是worker节点host或者IP，此处的配置为：

192.168.1.112

192.168.1.113

192.168.1.114

    至于spark-env.sh文件，可以配置如下属性：

(1) SPARK_MASTER_PORT:Master服务端口，默认为7077；

(2) SPARK_WORKER_CORES:每个Worker进程所需要的CPU核的数目；

(3) SPARK_WORKER_MEMORY:每个Worker进程所需要的内存大小；

(4) SPARK_WORKER_INSTANCES：每个Worker节点上运行Worker进程的数目；

(5) SPARK_MASTER_WEBUI_PORT：Master节点对应Web服务的端口；

(6)export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=192.168.1.117:2181,192.168.1.118:2181,192.168.1.119:2181 -Dspark.deploy.zookeeper.dir=/spark":用于指定Master的HA，依赖于zookeeper集群；

(7) export SPARK_JAVA_OPTS="-Dspark.cores.max=4"：用于限定每个提交的Spark Application的使用的CPU核的数目，因为缺省情况下提交的Application会使用所有集群中剩余的CPU Core。

    注意在Worker进程的CPU个数和内存大小的时候，要结合机器的实际硬件条件，如果一个Worker节点上的所有Worker进程需要的CPU总数目或者内存大小超过当前Worker节点的硬件条件，则Worker进程会启动失败。

将配置好的Spark文件拷贝至每个Spark集群的节点上的相同路径中。为方便使用spark-shell，可以在环境变量中配置上SPARK_HOME。

3.2.2 启动

    配置结束后，就该启动集群了。这里使用Master的HA方式，选取192.168.1.112、192.168.1.113节点作为Master，192.168.1.112、192.168.1.113、192.168.1.114节点上运行两个Worker进程。

首先在192.168.1.113节点上做此操作：

启动之后，可以查看当前节点的进程：

另外，为了保证Master的HA，在192.168.1.112节点上只启动Master：

192.168.1.112节点的进程为：

启动过后，通过Web页面查看集群的情况，这里访问的是：

http://192.168.1.113:8090/

再看standby节点192.168.1.112的web界面http://192.168.1.112:8090/

3.2.3 测试

    Spark的bin子目录中的spark-submit脚本是用于提交程序到集群中运行的工具，我们使用此工具做一个关于pi的计算。命令如下：

./bin/spark-submit --master spark://spark113:7077 \--class org.apache.spark.examples.SparkPi \--name Spark-Pi --executor-memory 400M \--driver-memory 512M \/home/hadoop/spark-1.0.0/examples/target/scala-2.10/spark-examples-1.0.0-hadoop2.0.0-cdh4.5.0.jar

    其中--master参数用于指定Master节点的URI，但是这里填的是Host，不是IP！

    任务启动之后，在Spark的Master的Web界面可以看到运行中的Application。

    任务运行结束之后，在Web界面中Completed Applications表格中会看到对应的结果。

同时，命令行中会打印出来运行的结果，如下所示：

4 spark-submit工具

    上面测试程序的提交都是使用的spark-submit脚本，其位于$SPARK_HOME/bin目录中，执行时需要传入的参数说明如下：

Usage: spark-submit [options] <app jar | python file> [app options]

    另外，在执行spark-submit.sh工具进行提交应用之前，可以使用如下方式提前定义好当前Spark Application所使用的CPU Core数目和内存大小：

SPARK_JAVA_OPTS="-Dspark.cores.max=2 -Dspark.executor.memory=600m" 
./bin/spark-submit --master spark://update113:7077 \--class org.apache.spark.examples.SparkPi 
…

…

5 Spark HistoryServer

    类似于Mapreduce的JobHistoryServer，Spark也有一个服务可以保存历史Application的运行记录。

    修改$SPARK_HOME/conf下的spark-defaults.conf文件(注意，修改后的配置文件在每个节点都要有)，其中可修改的配置属性为：

    此处的设置如下：

spark.eventLog.enabled  truespark.eventLog.dir      hdfs://yh/user/hadoop/sparklogsspark.yarn.historyServer.address    update113:18080

    设置完文件之后，进入sbin目录启动服务：

    运行完成的Application历史记录可以通过访问上面指定的HistoryServer地址查看，这里是http://192.168.1.113:18080/。

    无论运行时是本地模式，还是yarn-client、yarn-cluster，运行记录均可在此页面查看。

    并且程序运行时的环境变量、系统参数、各个阶段的耗时均可在此查看，很强大！

6 Spark可配置参数

Spark参数的配置可通过三种方式：SparkConf方式 > 命令行参数方式 >文件配置方式。

6.1 应用属性

6.2 运行环境变量

6.3 Shuffle操作相关属性

6.4 SparkUI相关属性

6.5 压缩和序列化相关属性

6.6 执行时相关属性

6.7 网络相关属性

6.8 调度相关属性

6.9 安全相关属性

6.10 SparkStreaming相关属性

6.11 Standalone模式特有属性

可以在文件conf/spark-env.sh中来设置此模式的特有相关属性：

(1)SPARK_MASTER_OPTS：配置master使用的属性

(2)SPARK_WORKER_OPTS：配置worker使用的属性

(3)SPARK_DAEMON_JAVA_OPTS：配置master和work都使用的属性

配置的时候，使用类似的语句：

export SPARK_MASTER_OPTS="-Dx1=y1 -Dx2=y2"

其中x代表属性，y代表属性值。

SPARK_MASTER_OPTS所支持的属性有：

SPARK_WORKER_OPTS所支持的属性有

SPARK_DAEMON_JAVA_OPTS所支持的属性有：

6.12 Spark on Yarn特有属性

7 示例配置

    主要的配置文件均位于$SPARK_HOME/conf中，包括slave、spark-env.sh、spark-defaults.conf文件等。

7.1 slave文件

192.168.1.112192.168.1.113192.168.1.114

7.2 spark-env.sh文件

export JAVA_HOME="/export/servers/jdk1.6.0_25" #yarn

export HADOOP_HOME=/home/hadoop/hadoop-2.0.0-cdh4.5.0export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

SPARK_EXECUTOR_INSTANCES=2SPARK_EXECUTOR_CORES=1SPARK_EXECUTOR_MEMORY=400M

SPARK_DRIVER_MEMORY=400M

SPARK_YARN_APP_NAME="Spark 1.0.0"#alone

SPARK_MASTER_WEBUI_PORT=8090SPARK_WORKER_MEMORY=400M

SPARK_WORKER_CORES=1SPARK_WORKER_INSTANCES=2#Master HA

export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=192.168.1.117:2181,192.168.1.118:2181,192.168.1.119:2181 -Dspark.deploy.zookeeper.dir=/spark"

7.3 spark-defaults.conf文件

#history server

spark.eventLog.enabled  truespark.eventLog.dir      hdfs://namespace/user/hadoop/sparklogsspark.yarn.historyServer.address    spark113:18080#shuffle

spark.shuffle.consolidateFiles true#task

spark.task.cpus 1spark.task.maxFailures 3#scheduler type

spark.scheduler.mode FAIR

#security

park.authenticate truespark.authenticate.secret hadoop

spark.core.connection.auth.wait.timeout 1500spark.ui.acls.enable truespark.ui.view.acls root,hadoop

#each executor used max memory

spark.executor.memory 400m

#spark on yarn

spark.yarn.applicationMaster.waitTries 5spark.yarn.submit.file.replication 3spark.yarn.preserve.staging.files falsespark.yarn.scheduler.heartbeat.interval-ms 5000#park standalone and on mesos

spark.cores.max 4

8 Spark SQL

    Spark支持Scala、Python等语言写的脚本直接在Spark环境执行，更重要的是支持对Hive语句进行包装后在Spark上运行。这就是Spark SQL。

8.1 相关配置

    配置的步骤比较简单，把Hive的配置文件hive-site.xml直接放置到$SPARK_HOME的conf路径下即可。如果是想在Spark集群本地执行SQL的话，每个对应的节点都要做同样的配置。

8.2 运行SQL

    启动bin目录下的spark-shell脚本，依次执行如下语句：

val sc: SparkContext

val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)

import hiveContext._

hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")

hql("LOAD DATA LOCAL INPATH ‘/examples /data.txt‘ INTO TABLE src")

hql("FROM src SELECT key, value").collect().foreach(println)

    上面的命令，分别是声明SparkContext对象，利用hql方法执行Hive的SQL语句，在执行SQL语句的过程中，可以通过Hive的Cli客户端进行查看相应操作的结果。

8.3 on yarn模式

    由于spark-shell脚本是在本地执行的，如果想放到Yarn上去执行的话，可以使用上面第4节中的spark-submit工具，这时候需要对需要输入的sql语句进行包装，将包装类打包成jar文件，再提交。

    包装类的代码如下：

 1 package spark; 2  3 import java.util.List; 4  5 import org.apache.spark.SparkConf; 6 import org.apache.spark.api.java.JavaSparkContext; 7 import org.apache.spark.sql.api.java.Row; 8 import org.apache.spark.sql.hive.api.java.JavaHiveContext; 9 10 /**11  * Description:12  * Author: ITScott@163.com13  * Date: 2014/7/1514  */15 public class SparkSQL {16 17     public static void main(String[] args) {18         if(args.length != 2){19             System.out.println("usage: <applicationName> <sql statments>");20             System.exit(1);21         }22 23         String applicationName = args[0];24         String sql = args[1];25 26         SparkConf conf = new SparkConf().setAppName(applicationName);27         JavaSparkContext sc = new JavaSparkContext(conf);28         JavaHiveContext hiveContext = new JavaHiveContext(sc);29         List<Row> results = hiveContext.hql(sql).collect();30 31         System.out.println("Sql is:" + sql + ", has been executed over.");32         System.out.println("The result size is " + results.size() + ", they are:");33         for(int i=0; i<results.size(); i++){34             System.out.println(results.get(i).toString());35         }36 37         System.out.println("Execute over ...");38         sc.stop();39         System.out.println("Stop over ...");40     }41 42 }

    将其打包成jar文件spark-0.0.1-SNAPSHOT.jar，再使用spark-submit工具进行任务的提交，命令如下：

./spark-submit \--class spark.SparkSQL \--master yarn-cluster \--num-executors 3 \--driver-memory 400m --executor-memory 400m --executor-cores 1 \--jars /home/hadoop/spark-1.0.0/examples/libs/spark-core_2.10-1.0.0.jar,/home/hadoop/spark-1.0.0/examples/libs/spark-hive_2.10-1.0.0.jar,/home/hadoop/spark-1.0.0/lib_managed/jars/datanucleus-api-jdo-3.2.1.jar,/home/hadoop/spark-1.0.0/lib_managed/jars/datanucleus-core-3.2.2.jar,/home/hadoop/spark-1.0.0/lib_managed/jars/datanucleus-rdbms-3.2.1.jar,/home/hadoop/hive-0.12.0/lib/mysql-connector-java-5.1.27-bin.jar--files /home/hadoop/spark-1.0.0/conf/hive-site.xml \/home/hadoop/spark-1.0.0/examples/libs/spark-0.0.1-SNAPSHOT.jar "hiveTest" "CREATE TABLE IF NOT EXISTS test4 (key INT, value STRING)"

其中，--master参数指定的是yarn-cluster模式，当然也可以使用yarn-client模式，至于区别，已经在上文说了；--class指定的是我们包装类的主类，见上文源码；--jars是依赖的四个jar包；--files是指定的hive-site.xml配置文件，提交到Yarn中的Application在执行的时候，需要把此配置文件分发到每个Executor上；最后的两个参数，一个是Application的名称，一个是运行的SQL语句。

    运行结束后，可以到Spark HistoryServer中查看运行结果。