Spark/RDD

迈不过友情╰ 2022-08-08 06:11 203阅读 0赞

每一个Spark应用(Job),都是由一个驱动程序(Driver)构成，它运行用户的main函数。

RDD，是一个元素集合，划分到集群的各个节点上，可以被并行操作。RDD是不可变更的。

两种类型的RDD，并行集合(Scala集合)和hadoop数据集。这两种类型的RDD都可以通过相同的方式进行操作。

### 驱动程序 ###

运行在Spark集群当中的每个驱动程序以各自独立的方式负责资源分配与任务调度工作。由于驱动程序充当着调度任务的执行者， 它需要最大程度与这些工作节点保持紧密 距离，从而避免网络延迟对执行效果造成的负面影响。

如果驱动程序停止工作，任务也将立即终止。如果集群管理器出现故障，新的任务则无法被提交，但是现有任务仍将继续保持执行。

### 并行集合 ###

并行集合是通过调用SparkContext的parallelize方法，在一个已经存在的Scala集合上创建的。集合的对象将会被拷贝，创建出一个可以被并行操作的分布式数据集。

一旦分布式数据集被创建好，它们将可以被并行操作。

并行集合的一个重要参数是slices，表示数据集切分的份数。一般情况下，Spark会尝试根据集群的状况，来自动设定slices的数目，也可以在创建RDD时指定，sc.parallelize(data,slices)。

Spark将会在集群上为每一份数据建立一个任务。

### Hadoop数据集 ###

Spark可以从存储在HDFS，或者Hadoop支持的其他文件系统(包括本地文件、HBase等)上的文件创建分布式数据集。Spark支持TextFile、SequenceFiles等。

对于TextFile ,sc.textFile(filepath)。

对于SequenceFiles， sc.sequenceFile\[K,V\]。其中K,V是文件中的key和values的类型，必须是Hadoop的Writable interface的子类。对于通用的Writable类型，Spark允许指定原生类型来代替，比如sequenceFile\[Int ,String\]将会自动读取IntWritable和Texts。

对于其他类型的Hadoop输入格式，可以使用SparkContext.hadoopRDD方法，可以接收任意类型的JobConf和输入格式类、键类型和值类型。

### RDD操作 ###

action在数据集上运行计算后，返回一个值给驱动程序Driver。

Spark上的所有transformation都是惰性的，它们并不会直接计算结果，只是记住应用到RDD上的这些转换动作。当发生一个要求返回结果(action)给Driver的动作时，这些转换才开始执行。

默认情况下，每一个转换过的RDD都会在它被执行一个动作时被重新计算。可以使用persist或者cache方法，持久化一个RDD在内存中。Spark也支持在磁盘上持久化数据集，或者在集群间复制数据集。

### RDD持久化 ###

当持久化一个RDD，每一个结点都将把它的计算分块结果保存在内存中。缓存是Spark构建迭代算法的关键。

Cache有容错机制，如果RDD的任一分区丢失了，通过使用原先创建它的转换操作，它将会被自动重算(不需要全部重算，只计算丢失的部分)。

每一个RDD都可以使用不同的保存级别进行保存，可以保存在硬盘、内存等。这些等级选择，通过将一个org.apache.spark.storage.StorageLevel对象传递给persist()方法进行确定。cache()使用默认的存储级别，StorageLevel.MEMORY\_ONLY。

### 存储级别 ###

MEMORY\_ONLY                   存储在JVM中。

MEMORY\_AND\_DISK