Spark基本概念扫盲全知道（持续更新修改中。。。。。）

谁借莪１个温暖的怀抱￠ 2022-05-24 02:50 190阅读 0赞

**RRD**

spark的核心

弹性分布式数据集 Resilient Distributed Datasets

RDD，包含一组分区列表（实际上是分区到block的映射，具体数据可以是分布式的存储在HDFS各个节点上）以及一组transformation或action算子。

RDD的五大特性

官网原文

1.A list of partitions

rdd里存的不是数据，只是分区到block块的映射，具体数据存在HDFS上或者内存里

正常一个block块对应一个partition，实际是一个split对应一个block块对应一个partition。但是有种特殊情况：

在Spark中实际上是没有读写文件的方法的，调用的是MapReduce的读写文件的方法，MR在读文件的时候先将数据分成一个个split切片，默认大小128m（hadoop2.0后，之前是64m）。特殊情况是假如一个文件128.1mb需要2个block来存储信息，第一个block块的尾部存储了最后一条记录的前半部分，第二个block块存了最后一条记录的后半段，但这时候在用MR读取文件这批数据会被切分成一个split。所以更准确的说法是 Rdd的partition数量=数据split的数量≈block数量

2. A function for computing each partition

在每一个partition上都有一个函数进行计算迭代

3 A list of dependencies on other RDDS

子rdd和父rdd之间是互相依赖的

依赖的好处

1 有利于计算的容错

2 有利于管道式的（pipeline）的计算

3 数据0丢失

4 Optionally, a Partitioner for key\_value RDDs

如果一个RDD是kv类型的，spark会提供一个partitioner分区器

分区器作用:决定了将map task 的计算结果写到哪一个磁盘小文件中

kv格式的RDD:如果这个RDD中存储的数据是二元组的话，那么这个rdd就是kv格式的rdd

一般用到了partitioner就一定发生了shuffle

5 Optionally，a list of preferred locations to compute echo

rdd会提供一系类的最佳计算位置

RDD有一个方法getPreferedLocation,能够获取到每一个分区的位置，task就可以发送到相应的位置去执行。体现了计算向数据移动，数据本地化的特点

\----------------------------------------------------------------------------------------------------------------------------------

**Application**

application（应用）就是一个spark-submit提交的一个程序。例如spark的example里面计算pi的sparkpi程序。application一般分为三部分：

从数据源（如HDFS）抽取数据形成RDD，RDD再经过一系列的transformation算子和action算子得到结果，最后输出带console或者外部HDFS

如何区分 transformation算子和action算子

如果一个算子的返回类型是RDD类型，那么这个算子是transformation类算子，否则就是action算子。action算子一般都伴随着shuffle发生

transformation算子是懒执行，需要action算子的触发。transformation算子类似于编剧一直在写剧本，编故事，单action导演说开始，才能真正执行

**提交application的方式：**

1 client提交方式

Driver进程是在客户端上启动

spark-submit --master spark://node005:7077 --class org.apache.spark.examples.SparkPi ../lib/spark-examples-1.6.0-hadoop2.6.0.jar 500

spark-submit --master --class jarPath args..

client这种提交方式可以看到一些调试信息，多用于测试场景

client会有网卡用量保证问题

2cluster提交方式

Driver进程是在集群的某一台Worker节点上启动

生产环境多用cluster提交方式，分散了网卡流量保证问题

spark-submit --master spark://node005:7077 --deploy-mode cluster --class org.apache.spark.examples.SparkPi ../lib/spark-examples-1.6.0-hadoop2.6.0.jar 500

spark on standalone  client端会启动一个ftp server 把需要的jar传进去，哪个worker需要自己拷贝，所以这种模式不依赖于HDFS

spark on yarn        使用spark集群，也要把hdfs起来，因为client会把所需要的jar上传到HDFS，等待被resourcemanager分配任务的nodemanager来下载jar包

\----------------------------------------------------------------------------------------------------------------------------------**Driver 进程**

就是一个任务调度的进程。

driver进程（jvm进程）有两个启动位置

client模式下，driver进程启动在提交application的机器上

cluster模式下，driver进程随机启动在集群里的一个机器上

作用 1 负责分发任务到worker进程所在的结点

2 获取数据的位置信息

3监控task得执行情况

4 重试task得执行

5  回收每一个task得计算结果，存储到HDFS中，这样不能造成om

\----------------------------------------------------------------------------------------------------------------------------------

**worker node**

负责本机资源调度

worker node 就是集群中任何一个可以运行spark程序的节点，可以在上面启动executor进程

\----------------------------------------------------------------------------------------------------------------------------------

**master  node**

作用 1 接受worker的注册消息

2 管理每个worker进程，进而管理整个cluster的资源情况

\----------------------------------------------------------------------------------------------------------------------------------

**job**

spark里的job不同于MapReduce里的job，mr里一般指一个map job 或一个reduce job。在spark里一个action算子就是一个job。

\----------------------------------------------------------------------------------------------------------------------------------

**Stage**

\----------------------------------------------------------------------------------------------------------------------------------

**Executor  **

是一个jvm进程，最低配置1g内存，一个核,主要负责执行task

\----------------------------------------------------------------------------------------------------------------------------------

**RRD的持久化**

先说说为啥要持久化

一个application执行如下

HDFS----RDD1------RDD2-----RDD3:   ---------action1算子

---------action2算子

----------action3算子

从rdd1 经历了一系列的transformation算子得到了rdd3，基于rdd3这一个结果分别执行了action1，2, 3三个action算子

因为rdd3之前的算子都是懒执行算子，所以action1基于rdd3执行完了，到了执行action2的时候也要基于rdd3，但是问题来了，rdd3在action1执行完了后并没存，action2,3要是想基于rdd3计算还需要重新执行。因此我们需要把rdd3这个数据集持久化，便于action2,3的复用。

cache算子：会将RDD的计算结果持久化到内存里

persist算子：会根据用户指定的持久化级别，将数据放到不同的地方

持久化的算子是一个懒执行算子。一般放到action算子之前。在演戏之前把持久化算子写到剧本上，让下一个action算子执行。

在action1中实际上没有使用到持久化的数据，但是在接下来的action中就可以使用到持久化的信息了，如果你的application中只有一个action1，通过持久化这一手段无法提高效率。

注意点：

1 cache或者persist算子返回的结果必须赋值给一个变量，在接下来的job中直接使用这个变量，那么就使用到了这个持久化的数据

2 持久化的算子都是懒执行算子，需要有一个action类的算子出发执行，但是持久化的算子后面不能立即紧跟action类算子

val abc =rdd.cache().count()  --------------错

val  abc=rdd.cache()

rdd.count

persist算子可以执行的持久化级别： 红色为常用的

1. val DISK\_ONLY = new StorageLevel(true, false, false, false)

持久化到硬盘上

2.  val DISK\_ONLY\_2 = new StorageLevel(true, false, false, false, 2)

持久化到硬盘上且有两个备份

3. val MEMORY\_ONLY = new StorageLevel(false, true, false, true)

只是将数据持久化到内存中，和cache一样，cache是个简化版

4.  val MEMORY\_ONLY\_2 = new StorageLevel(false, true, false, true, 2)

将数据持久化到内存中,且有两个备份

5. val MEMORY\_ONLY\_SER = new StorageLevel(false, true, false, false)

将序列化的数据持久化到内存中

6. val MEMORY\_ONLY\_SER\_2 = new StorageLevel(false, true, false, false, 2)

将序列化的数据持久化到内存中，且有两个备份

7. val MEMORY\_AND\_DISK = new StorageLevel(true, true, false, true)

将数据持久化到内存中，内存放不下才放到硬盘中

8.  val MEMORY\_AND\_DISK\_2 = new StorageLevel(true, true, false, true, 2)

将数据持久化到内存中，内存放不下才放到硬盘中，且有两个备份，有可能导致效率降低

9.  val MEMORY\_AND\_DISK\_SER = new StorageLevel(true, true, false, false)

将序列化的数据持久化到内存中，内存放不下才放到硬盘中

10.  val MEMORY\_AND\_DISK\_SER\_2 = new StorageLevel(true, true, false, false, 2)

将序列化的数据持久化到内存中，内存放不下才放到硬盘中，且有两个备份

11.  val OFF\_HEAP = new StorageLevel(false, false, true, false)

存到堆外内存，需要spark和Tachyon整合。 Tachyon是一个内存分布式文件系统

12.  val NONE = new StorageLevel(false, false, false, false)

不持久化

如果需要持久化的数据非常大，而你还选择了memory\_only，那么内存中能存多少存多少，存不了的要是后面用到了再根据rdd之间的依赖关系重新算。不会造成内存溢出。如果持久化的数据不全，会找RDD的父RDD重新计算不在内存中的数据。在spark\_driver进程里会有一个对象cacheManager记录哪些数据存了，哪些没有存。

持久化的最小单位是Partition,要不都存，要不都不存。