Spark系列(二)执行Spark程序

待我称王封你为后i 2022-05-21 11:56 247阅读 0赞

## 一、执行第一个spark程序（standalone） ##

**该算法是利用蒙特·卡罗算法求PI**

/opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-submit \
    --class org.apache.spark.examples.SparkPi \ --master spark://hadoop0:7077 \ --executor-memory 1G \ --total-executor-cores 2 \ /opt/module/spark-2.1.1-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.1.1.jar \

**参数说明：**  
**–master spark://hadoop0:7077**  
指定Master的地址

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

**–total-executor-cores 2**  
指定每个executor使用的cup核数为2个

控制台日志出现如下，即为成功：

pi is roughly 3.14........

--------------------

## 二、Spark应用提交 ##

一旦打包好,就可以使用bin/spark-submit脚本启动应用了. 这个脚本负责设置spark使用的classpath和依赖,支持不同类型的集群管理器和发布模式:

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

**一些常用选项:**

1）–class: 你的应用的启动类 (如 org.apache.spark.examples.SparkPi)  
2）–master: 集群的master URL (如 spark://23.195.26.187:7077)  
3）–deploy-mode: 是否发布你的驱动到worker节点(cluster) 或者作为一个本地客户端 (client) (default: client)\*

4）–conf: 任意的Spark配置属性， 格式key=value. 如果值包含空格，可以加引号“key=value”. 缺省的Spark配置

5）application-jar: 打包好的应用jar,包含依赖. 这个URL在集群中全局可见。 比如hdfs:// 共享存储系统， 如果是 file:// path， 那么所有的节点的path都包含同样的jar.

6）application-arguments: 传给main()方法的参数

**查看Spark-submit全部参数：**

/bin/spark-submit

**注：我们可以再IDEA里用Scala编写程序，然后打包成jar，再提交到Spark集群运行**

--------------------

## 三、启动Spark Shell ##

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

### 3.1 启动Spark Shell ###

/opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-shell \
    --master spark://hadoop0:7077 \
    --executor-memory 2g \
    --total-executor-cores 2

**注意:** 可能会出现内存不足的错误，在[http://hadoop0:8080][http_hadoop0_8080]查看已完成的任务，点击任务，查看State=KILLED的，点击logs里的stderr，如在日志里发现CoarseGrainedExecutorBackend: RECEIVED SIGNAL TERM  
则虚拟机内存可能不足，我之前4G内存出现上述错误，后重新分配为8G内存，就OK了。

**注意：**

>  *  如果启动spark shell时没有指定master地址，但是也可以正常启动spark shell和执行spark shell中的程序，其实是启动了spark的local模式，该模式仅在本机启动一个进程，没有与集群建立联系。
>  *  前面的错误，本人不调整内存，直接/opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-shell 也能正常启动。如果你想以Spark集群启动，机器内存一定得够。

**Spark Shell中已经默认将SparkContext类初始化为对象sc。用户代码如果需要用到，则直接应用sc即可**

### 3.2 在Spark shell中编写WordCount程序 ###

**1）首先启动hdfs，并上传WC文件**

将Spark目录下的RELEASE文件上传一个文件到hdfs://hadoop0:9000/RELEASE

$ hadoop fs -put ./RELEASE /

**2）在Spark shell中用scala语言编写spark程序**

sc.textFile("hdfs://hadoop0:9000/RELEASE").flatMap(_.split(" "))
    .map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://hadoop1:9000/out")

3）使用hdfs命令查看结果

[root@hadoop1 spark-2.1.1-bin-hadoop2.7]# hadoop fs -cat /out/p*
    (-Psparkr,1)
    (Build,1)
    (built,1)
    (2.1.1,1)
    (-Phive-thriftserver,1)
    (-Pmesos,1)
    (2.7.3,1)
    (-Phadoop-2.7,1)
    (Spark,1)
    (-Pyarn,1)
    (-DzincPort=3036,1)
    (flags:,1)
    (for,1)
    (-Phive,1)
    (Hadoop,1)
    [root@hadoop1 spark-2.1.1-bin-hadoop2.7]#

**说明：**

> sc是SparkContext对象，该对象时提交spark程序的入口，  
> textFile(hdfs://hadoop0:9000/RELEASE)是hdfs中读取数据，  
> flatMap(\_.split(” “))先map在压平，  
> map((\_,1))将单词和1构成元组，  
> reduceByKey(*+*)按照key进行reduce，并将value累加，  
> saveAsTextFile(“hdfs:// hadoop0:9000/out”)将结果写入到hdfs中。

[http_hadoop0_8080]: http://hadoop0:8080