spark提供了三種叢集模式：standalone、yarn以及Mesos，三種模式裡面standalone模式是一個基礎，本篇先從standalone模式講解一個基礎的spark叢集搭建過程，並且基於這個叢集我們再介紹一下spark-shell的使用、spark提供的例子如何執行，以及開發一個簡單的例子通過任務提交的方式執行起來。

Spark叢集搭建(standalone模式)

安裝

我們以兩臺機器作為叢集搭建，其中spark-master作為master，同時也作為slave，名稱為spark-slave0，另外一臺機器作為slava，名稱為spark-slave1，他們的網路架構如下圖所示：

步驟2：tar -xvf spark-2.0.1-bin-hadoop2.6

步驟3：cd  ./spark-2.0.1-bin-hadoop2.6/conf

步驟4：cp slaves.template slaves

步驟5：cp spark-env.sh.template spark-env.sh

步驟6：修改啟動環境變數指令碼vim spark-env.sh，新增

export JAVA_HOME="/usr/local/java"
export SCALA_HOME="/usr/local/scala"
export SPARK_HOME="/data/spark-2.0.1-bin-hadoop2.6"
export SPARK_MASTER_HOST=spark-master
export SPARK_WORKER_CORES=1
export SPARK_WORKER_INSTANCES=1
export SPARK_MASTER_PORT=7077
export SPARK_WORKER_MEMORY=1g
export MASTER=spark://${SPARK_MASTER_IP}:${SPARK_MASTER_PORT}
 

步驟7：修改slaves配置,vim slaves，新增

spark-slave0
spark-slave1

步驟8：系統環境變數配置(假設系統已安裝jdk及scala),vim /etc/profile,新增

export JAVA_HOME="/usr/local/java"
export SPARK_HOME="/data/spark-2.0.1-bin-hadoop2.6"
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib

步驟9：source /etc/profile

步驟10：修改host配置 vim /etc/hosts 新增

A-host-ip spark-master
A-host-ip spark-slave0
B-host-ip spark-slave1

注：此處請務必注意，A-host-ip一定要寫的是A機器的IP，而不是127.0.0.1,如果是寫了後者，master啟動後會只監聽本地迴環地址127.0.0.1，而導致其他slave例如B機器連線被拒絕。

步驟11：B機器配置按照A機器配置完全一樣複製一份即可。

配置ssh叢集間無密登入

步驟1：確保A,B機器之間ping通

步驟2：關閉防火牆限制

2.1、重啟後生效

開啟： chkconfigiptables on

關閉： chkconfigiptables off

2.2、即時生效，重啟後失效

開啟： service iptablesstart

關閉： service iptables stop

步驟3：使用root使用者在各個節點建立金鑰對

[[email protected] ~]$ ssh-keygen -t rsa
[[email protected] ~]$ ssh-keygen -t rsa
如下所示:
[[email protected] conf]# ssh-keygen -t rsa
Generating public/private rsa key pair.
Enter file in which to save the key (/root/.ssh/id_rsa): 
/root/.ssh/id_rsa already exists.
Overwrite (y/n)? y
Enter passphrase (empty for no passphrase): 
Enter same passphrase again: 
Your identification has been saved in /root/.ssh/id_rsa.
Your public key has been saved in /root/.ssh/id_rsa.pub.
The key fingerprint is:
c5:54:cf:ee:3b:43:19:9d:63:72:11:83:d5:84:b6:ad [email protected]
The key's randomart image is:
+--[ RSA 2048]----+
|          ... o*+|
|         o   +o.o|
|          o  .o+o|
|         .   oo=o|
|        S     =+.|
|             .E  |
|             ..  |
|              o. |
|              .o |
+-----------------+

這樣執行之後就在/root/.ssh/目錄下生成了一對ssh的公私鑰。id_rsa 以及id_rsa.pub

步驟4：切換到spark-master節點合併節點公鑰。

[[email protected]]$ ssh spark-slave1cat /root/.ssh/id_rsa.pub>>authorized_keys

[[email protected]]$ ssh spark-slave0cat /root/.ssh/id_rsa.pub>>authorized_keys

兩臺slave的公鑰在master上合併完成。

注：本例採用的是root賬號執行，如果你採用的是hadoop賬號執行以上所有安裝操作就需要修改authorized_keys檔案的屬性（.ssh目錄為700，authorized_keys檔案為600，用chmod命令修改），不然ssh登入的時候還需要密碼。

步驟5：驗證一下ssh無密是否成功：[[email protected] .ssh]$ ssh spark-slave1 date

啟動standalone叢集

cd /data/spark-2.0.1-bin-hadoop2.6/sbin

./start-all.sh

啟動之後，master有一個webui介面，訪問如下地址：http://A:8080如下

Spark-shell使用

Spark-shell提供了基於scala語言的指令碼執行環境，在啟動的時候會自動建立一個名為sc的sparkContext上下文，具體啟動方式

cd /data/spark-2.0.1-bin-hadoop2.6/bin

./spark-shell

另外spark-shell還可以提交任務來允許，如下幾種模式：

本地master4核執行

$ ./bin/spark-shell --master local[4]

新增jar包載入到classpath執行

$ ./bin/spark-shell --master local[4] --jars code.jar

$ ./bin/spark-shell --master local[4] --packages "org.example:example:0.1"

任務提交

我們首先編寫一個RDD(spark的儲存檔案型別，後續章節我們重點介紹下RDD的結構)測試小程式，通過這個小程式打包jar提交到我們剛才搭建的環境中執行。Spark任務提交是通過spark-submit來提交的，一個典型的spark-submit的引數解釋如下：

./bin/spark-submit \
  --class <main-class> \
  --master <master-url> \
  --deploy-mode <deploy-mode> \
  --conf <key>=<value> \
  ... # other options
  <application-jar> \
  [application-arguments]

--class 要執行的類

--master sparkmaster的地址，一般的寫法是spark://host:port

--deploy-model 部署模式叢集執行cluster：叢集各節點分散式執行client：本地執行

請注意cluster執行的時候master一定要寫它的rest地址，這個可以再master的webui上檢視具體地址。

--conf spark的配置引數，鍵值對形式

Application-jar 所需要執行的jar包路徑

我們編寫的小例項：

Pom.xml檔案如下

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>spark.test</groupId>
	<artifactId>rdd</artifactId>
	<version>1.0</version>
	<dependencies>
		<dependency> <!-- Spark dependency -->
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-core_2.10</artifactId>
			<version>1.6.1</version>
			<scope>provided</scope>
		</dependency>

		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-mllib_2.10</artifactId>
			<version>1.6.1</version>
			<scope>provided</scope>
		</dependency>

		<dependency>
			<groupId>commons-cli</groupId>
			<artifactId>commons-cli</artifactId>
			<version>1.2</version>
		</dependency>
	</dependencies>
</project>

package rdd;

import java.util.Arrays;
import java.util.HashMap;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;

/**
 * 
 * @author flyking
 *
 */
public class TestRDD {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("spark://master:port").setAppName("SimpleRDD");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1, 2, 3 ,3), 2);
        
        System.out.println("rdd collect" + rdd.collect());
        System.out.println("rdd count" + rdd.count());
        System.out.println("rdd countByValue" + rdd.countByValue());
        System.out.println("rdd take" + rdd.take(2));
        System.out.println("rdd top" + rdd.top(2));
        System.out.println("rdd takeOrdered" + rdd.takeOrdered(2));
        System.out.println("rdd map "+ rdd.map(new Function<Integer, HashMap<String,Integer>>() {
			public HashMap<String,Integer> call(Integer v1) throws Exception {
				return (HashMap<String, Integer>) new HashMap().put(String.valueOf(v1), v1);
			}
		}));
        System.out.println("rdd reduce" + rdd.reduce(new Function2<Integer, Integer, Integer>() {
			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1+v2;
			}
		}));
	}
}

把檔案上傳到master伺服器/data/rdd-1.0.jar,我們採取本地模式提交任務執行

cd /data/spark-2.0.1-bin-hadoop2.6/bin
./spark-submit --class rdd.TestRDD --master local[2] --executor-memory 2G --num-executors 5 /data/rdd-1.0.jar 

執行後的結果如下：

rdd collect[1, 2, 3, 3]
rdd count4
rdd countByValue{2=1, 1=1, 3=2}
rdd take[1, 2]
rdd top[3, 3]
rdd takeOrdered[1, 2]
rdd map MapPartitionsRDD[6] at map at TestRDD.java:26
rdd reduce9