Spark之本地模式与集群模式

一时失言乱红尘 2022-05-25 04:08 173阅读 0赞

### Spark之本地模式与集群模式 ###

#### 1.spark-shell的本地模式和集群模式 ####

**1.1 local本地模式**

直接启动spark-shell命令窗口

脚本启动后，会生成一个`SparkContext`的上下文对象`sc`。并且启动的是本地模式(local)。如图：

![image][]

1.1.1 加载本地数据

sc.textFile("file:///home/hadoop/words.txt").flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).sortBy(_._2,true).collect()

![image][image 1]

1.1.2 加载hdfs数据

# 因为在spark-env.sh中配置了HADOOP_CONF_DIR目录，所以默认使用hdfs文件系统。
    # /spark/words.txt：表示hdfs文件系统目录
    sc.textFile("/spark/words.txt").flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).sortBy(_._2,true).collect()

![image][image 2]

**1.2 集群模式**

1.2.1 启动集群模式的`spark-shell`窗口

spark-shell --master spark://server01:7077 --total-executor-cores 3 --executor-memory 1g

参数介绍：

--master spark://server01:7077:指定master进程的机器 
    --total-executor-cores 3:指定executor的核数（worker数量） 
    --executor-memory 1g:指定executor执行的内存大小

![image][image 3]

1.2.2 代码执行

sc.textFile("/spark/words.txt").flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).sortBy(_._2).collect()

![image][image 4]

1.2.3 结果查看

![image][image 5]

通过集群模式提交的任务，在web页面上是有展示的

**本地模式（local）和集群模式的区别**

1.本地模式不运行在集群上，运行在当前执行的机器上

2.本地模式的任务不会在web页面显示

3.本地模式是采用线程来模拟集群的worker进程

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

#### 2. scala api实现的本地模式和集群模式 ####

**3.1 local本地模式**

object WordCountLocal {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("wordcount").setMaster("local[2]")
        val sc = new SparkContext(conf)
        sc.textFile("file:///d:/a.txt").flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _).sortBy(_._2).saveAsTextFile("file:///d:/out")
      }
    }

因为本地安装了spark，所以可以直接在本地运行，在本地运行。

`setMaster("local[2]")`中`local`表示本地运行,`[2]`表示是使用2个线程。

3.2 **生成jar提交到集群**

代码：

object WordCountMaster {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("WordCountMaster")
        val sparkContext = new SparkContext(conf)
        sparkContext.textFile(args(0)).flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _).sortBy(_._2).saveAsTextFile(args(1))
      }
    }

提交到集群：

spark-submit \
    --master spark://server01:7077 \ --executor-memory 1g \ --total-executor-cores 2 \ --class com.yundoku.spark.WordCountMaster \ /home/hadoop/sparkscalawordcount.jar \
    /spark/words.txt \
    /spark/scala_wordcount_out

参数讲解：

--master:指定集群的master 
    --executor-memory:executor的内存大小（worker） 
    --total-executor-cores 2:executor核数为2（worker的执行个数） 
    --class com.yundoku.spark.WordCountMaster:包含main的类，程序的入口 /home/hadoop/sparkscalawordcount.jar :jar文件
     /spark/words.txt:输入参数1
     /spark/scala_wordcount_out：输入参数2

运行结果：

![image][image 6]

这里会生成`part-000000`和`part-000001`2个结果文件，表示有2个分区。

原因是，在spark的读取文件时默认是使用的最小分区为2

![image][image 7]

defaultMinPartitions的值初始化，如下图所示

![image][image 8]

[image]: https://raw.githubusercontent.com/zhang3550545/image_center/master/image-2018/spark-2.1.png
[image 1]: https://raw.githubusercontent.com/zhang3550545/image_center/master/image-2018/spark-2.2.png
[image 2]: https://raw.githubusercontent.com/zhang3550545/image_center/master/image-2018/spark-2.3.png
[image 3]: https://raw.githubusercontent.com/zhang3550545/image_center/master/image-2018/spark-2.4.png
[image 4]: https://raw.githubusercontent.com/zhang3550545/image_center/master/image-2018/spark-2.5.png
[image 5]: /images/20220525/80d0bfda91d94e13ba69498df2a91b9c.png
[image 6]: https://raw.githubusercontent.com/zhang3550545/image_center/master/image-2018/spark-2.7.png
[image 7]: https://raw.githubusercontent.com/zhang3550545/image_center/master/image-2018/spark-2.8.png
[image 8]: https://raw.githubusercontent.com/zhang3550545/image_center/master/image-2018/spark-2.9.png