一、执行第一个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。