Spark编程基础

目次
一，Spark计划与运行原理
1，Spark简介
2，Spark与Hadoop对比
二，Spark运行架构
1，RDD计划与运行原理
方法1.parallelize()
方法2.makeRDD()
方法3.通过HDFS文件创建 RDD
方法4.通过 Linux 当地文件创建 RDD
2,RDD方法归纳
1.使用map()方法转换数据​
2.使用 sortBy（）方法进行排序 ​
 3.使用flatMap（）方法转换数据​
4.使用take()方法查询某几个值 ​
 5.使用union()方法合并多个RDD​
6.使用distinct()方法进行去重 ​
3， 使用简单的集合操作​
 (1)intersection()方法​
 (2)subtract()方法​
(3)cartesian()方法 ​
任务实现： 
三，Spark快速上手
 1，创建Maven项目
（1）增加Scala插件​
（2）增加依靠关系 
 （3）WordCount
（4）非常处理处罚
四，Spark运行情况 
1，Local模式
上传并解压缩文件
启动local情况
下令行工具 
退出当地模式
提交应用
五，Spark运行架构
1，运行架构
2，焦点概念
（1）Executor与Core

---

一，Spark计划与运行原理

1，Spark简介

Spark 是一种基于内存的快速、通用、可扩展的大数据分析计算引擎
特点：运行速度快、容易使用、通用性、运行模式多样
2，Spark与Hadoop对比

Hadoop存在的缺点：
表达本领有限 ，磁盘IO开销大 ，延长高
Spark长处：
编程模子更机动，迭代运算服从更高，任务调度机制更优
二，Spark运行架构

1，RDD计划与运行原理

创建RDD
方法1.parallelize()

parallelizeO方法有两个输人参数，说明如下:
（1）要转化的集合：必须是 Seq集合。Seq 表现序列，指的是一类具有一定长度的、可迭代访问的对象，其中每个数据元素均带有一个从0开始的、固定的索引。
（2）分区数。若不设分区数，则RDD 的分区数默认为该程序分配到的资源的 CPU焦点数。
通过 parallelizeO方法用一个数组的数据创建RDD，并设置分区数为4，创建后查看该 RDD 的分区数
方法2.makeRDD()

makeRDD0方法有两种使用方式，第一种使用方式与 parallelize0方法划一；第二种方式是通过接收一个 Seq［（T,Seq［String］）］参数类型创建 RDD。第二种方式天生的RDD中保存的是T的值，Seq［String］部分的数据会按照 Seqf（T,Seq［String］）的次序存放到各个分区中，一个 Seq［Stringl对应存放至一个分区，并为数据提供位置信息，通过preferredLocations0方法可以根据位置信息查看每一个分区的值。调用 makeRDD0时不可以直接指定 RDD 的分区个数，分区的个数与 Seq［String］参数的个数是保持划一的，使用 makeRDD0方法创建 RDD，并根据位置信息查看每一个分区的值
方法3.通过HDFS文件创建 RDD

这种方式较为简单和常用，直接通过 textFile（）方法读取 HDFS文件的位置即可。
在HDFS 的/user/toot 目次下有一个文件test.txt，读取该文件创建一个 RDD
方法4.通过 Linux 当地文件创建 RDD

当地文件的读取也是通过 sc.textFile（"路径"）的方法实现的，在路径前面加上“file://”表现从Linux 当地文件系统读取。在 IntelliJIDEA 开辟情况中可以直接读取当地文件；但在 spark-shell 中，要求在所有节点的相同位置保存该文件才可以读取它，例如，在Linux的/opt 目次下创建一个文件 test.txt，恣意输入4行数据并保存，将 test.txt 文件远程传输至所有节点的/opt 目次下，才可以读取文件 test.txt。读取 test.txt 文件，并且统计文件的数据行数
2,RDD方法归纳

1.使用map()方法转换数据

2.使用 sortBy（）方法进行排序 

 3.使用flatMap（）方法转换数据

4.使用take()方法查询某几个值 

 5.使用union()方法合并多个RDD

6.使用distinct()方法进行去重 

3， 使用简单的集合操作

 (1)intersection()方法

 (2)subtract()方法

(3)cartesian()方法 

任务实现： 

三，Spark快速上手

 1，创建Maven项目

（1）增加Scala插件

（2）增加依靠关系 

    修改Maven项目中的POM文件，增加Spark框架的依靠关系

1. <dependencies>
3.         <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
5.         <dependency>
7.             <groupId>org.apache.spark</groupId>
9.             <artifactId>spark-core_2.12</artifactId>
11.             <version>2.4.5</version>
13.         </dependency>
15.     </dependencies>
17.     <build>
19.         <plugins>
21.             <!-- 该插件用于将 Scala 代码编译成 class 文件 -->
23.             <plugin>
25.                 <groupId>net.alchim31.maven</groupId>
27.                 <artifactId>scala-maven-plugin</artifactId>
29.                 <version>3.2.2</version>
31.                 <executions>
33.                     <execution>
35.                         <!-- 声明绑定到 maven 的 compile 阶段 -->
37.                         <goals>
39.                             <goal>testCompile</goal>
41.                         </goals>
43.                     </execution>
45.                 </executions>
47.             </plugin>
49.             <plugin>
51.                 <groupId>org.apache.maven.plugins</groupId>
53.                 <artifactId>maven-assembly-plugin</artifactId>
55.                 <version>3.1.0</version>
57.                 <configuration>
59.                     <descriptorRefs>
61.                         <descriptorRef>jar-with-dependencies</descriptorRef>
63.                     </descriptorRefs>
65.                 </configuration>
67.                 <executions>
69.                     <execution>
71.                         <id>make-assembly</id>
73.                         <phase>package</phase>
75.                         <goals>
77.                             <goal>single</goal>
79.                         </goals>
81.                     </execution>
83.                 </executions>
85.             </plugin>
87.         </plugins>
89.     </build>

复制代码

 （3）WordCount

   为了能直观地感受Spark框架的效果，接下来我们实现一个大数据学科中最常见的教学案例         WordCount

1. /**
2. * spark实现单词计数
3. */
5. object WordCountSpark {
7.   def main(args: Array[String]): Unit = {
9.     //创建spark运行配置对象
11.     val spark: SparkConf = new SparkConf()
13.       .setMaster("local[*]")
15.       .setAppName("WordCountSparkApps")
17.     //创建spark上下文对象
19.     val sc: SparkContext = new SparkContext(spark)
21.     //读文件数据
23.     val wordsRDD: RDD[String] = sc.textFile("data/word.txt")
25.     //讲文件中的数据进行分词
27.     val word: RDD[String] = wordsRDD.flatMap(_.split(","))
29.     //转换数据结构word ---->(word,1)
31.     val word2: RDD[(String, Int)] = word.map((_, 1))
33.     //将转换结构后的数据按照相同的单词进行分组聚合
35.     val word2CountRDD: RDD[(String, Int)] = word2.reduceByKey(_ + _)
37.     //将数据聚合结果采集到内存中
39.     val word2Count: Array[(String, Int)] = word2CountRDD.collect()
41.     //打印结果
43.     word2Count.foreach(println)
45.     //关闭spark连接
47.     sc.stop()
49.   }
51. }

复制代码

（4）非常处理处罚

假如本机操作系统是Windows，在程序中使用了Hadoop相关的东西，比如写入文件到HDFS，则会碰到如下非常：

出现这个问题的缘故原由，并不是程序的错误，而是windows系统用到了hadoop相关的服务，办理办法是通过设置关联到windows的系统依靠就可以了 

在IDEA中设置RunConfiguration，添加HADOOP_HOME变量大概在windows上设置情况变量： 

四，Spark运行情况 

1，Local模式

上传并解压缩文件

（1）上传文件至/usr/local/packages中

（2）解压缩到指定目次 
[root@master local]# tar -zxvf spark-2.4.5-bin-hadoop2.6.tgz -C /usr/local/soft/
（3）重定名 
[root@master soft]# mv spark-2.4.5-bin-hadoop2.6/ spark-local
启动local情况

（1）进入解压缩（spark-local）目次

（2）启动成功后，可以输入网址进行Web UI监控页面访问 

下令行工具 

sc.textFile("data/word.txt").flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).collect
退出当地模式

scala> :quit 
提交应用

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master local[1] \
./examples/jars/spark-examples_2.11-2.4.5.jar \
10
五，Spark运行架构

1，运行架构

Spark框架的焦点是一个计算引擎，团体来说，它采用了标准master-slave的结构。如下图所示，它展示了一个 Spark 实行时的根本结构。图形中的Driver表现 master，负责管理整个集群中的作业任务调度。图形中的Executor则是 slave，负责实际实行任务。

2，焦点概念

（1）Executor与Core

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。