一、执行第一个Spark程序

1、执行程序

我们执行一下Spark自带的一个例子，利用蒙特·卡罗算法求PI：

启动Spark集群后，可以在集群的任何一台机器上执行一下命令：

/home/spark/spark-1.6.1-bin-hadoop2.6/bin/spark-submit \

--class org.apache.spark.examples.SparkPi \

--master spark://master:7077 \

--executor-memory 1G \

--total-executor-cores 2 \

/home/spark/spark-1.6.1-bin-hadoop2.6/lib/spark-examples-1.6.1-hadoop2.6.0.jar \

100

在执行过程中bash上的信息:

执行完成bash上的信息:

执行过程中WebUI上的信息:

执行完以后WebUI上的信息:

2、命令解析

/home/spark/spark-1.6.1-bin-hadoop2.6/bin/spark-submit \

①spark-submit：提交任务，它是一个Driver，至于它的实现细节后续会有介绍

--class org.apache.spark.examples.SparkPi \

②—class 指定任务的类名（使用反射调用该类的main方法）

--master spark://master:7077 \

③—master 指定集群Master的地址

--executor-memory 1G \

④—executor 指定为每个executor分配的内存大小

--total-executor-cores 2 \

⑤—total-executor 指定分配给所有executor总的处理器核数

（这里先说一下executor是Worker启动的子进程，executor负责执行任务，其细节以后会介绍的。）

/home/spark/spark-1.6.1-bin-hadoop2.6/lib/spark-examples-1.6.1-hadoop2.6.0.jar \

⑥指定任务的jar包地址

100

⑦任务的类的main方法的参数

二、使用 spark-shell

spark-shell是Spark自带的交互式Shell程序，方便用户进行交互式编程，用户可以在该命令行下用scala编写spark程序，spark-shell也是一个Driver。

1、启动 spark-shell

/home/spark/spark-1.6.1-bin-hadoop2.6/bin/spark-shell \

--master spark://master:7077 \

--executor-memory 1g \

--total-executor-cores 2

这个命令的解析其实和上面的spark-submit是一样的，我在这再啰嗦一遍：

参数说明：

--master spark://master:7077 指定Master的地址

--executor-memory 1g 指定每个executor可用内存为1G

--total-executor-cores 2 指定所有executor总的处理器核数为2

还有一点需要非常注意：

如果启动 spark shell 时没有指定master地址，但是也可以正常启动 spark shell 和执行 spark shell 中的程序，其实是启动了spark的local模式，该模式仅在本机启动一个进程，没有与集群建立联系。

还要说明一点：

spark shell 中已经默认将SparkContext类初始化为对象sc。用户代码如果需要用到，则直接应用 sc 即可。SparkContext是Spark集群的入口，Driver只有初始化了SparkContext才可以向Spark集群提交任务，所以这个SparkContext和重要，以后我们会详细介绍整个SparkContext的初始化流程的，现在可以先记住SparkContext是集群的入口，就像Spring中的ApplicationContext一样。

2、在spark shell中编写WordCount程序

（1）首先启动hdfs

（2）向hdfs上传文件words.txt 到 hdfs://hadoop01:9000/spark/words.txt

words.txt的内容：

hello tom
hello jim
hello tom and kitty

（3）在spark shell 中用scala语言编写spark程序：

sc.textFile("hdfs://hadoop01:9000/spark/words.txt").flatMap(_.split(" "))

.map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://hadoop01:9000/spark/out")

（4）使用hdfs命令查看结果：

hdfs dfs -cat hdfs://hadoop01:9000/spark/out/part*

(jim,1)
(tom,2)
(hello,3)
(and,1)
(kitty,1)

（5）程序简单说明：

sc.textFile("hdfs://hadoop01:9000/spark/words.txt") 从hdfs中读取数据

flatMap(_.split(" ")) 先map再压平

map((_,1)) 将单词和1构成元组

reduceByKey(_+_) 按照key进行reduce，并将value累加

saveAsTextFile("hdfs://hadoop01:9000/spark/out") 将结果写入到hdfs中

最后：

也可以使用 IDEA 编写完一个程序后打包，使用spark-submit方式提交到集群，在这里我就不写了。一定要注意spark-submit的配置命令不要出错，还要注意自己的程序需要的参数的正确，不要忘了起hdfs。