Spark大数据分析 课件 第4章 弹性分布式数据集

RDD概述目录/Contents01

RDD概述RDD概述01RDD概述RDD（ResilientDistributedDataset）叫做分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。在Spark中，对数据的所有操作不外乎创建RDD、转化已有RDD以及调用RDD操作进行求值。每个RDD都被分为多个分区，这些分区运行在集群中的不同节点上。RDD可以包含Python、Java、Scala中任意类型的对象，甚至可以包含用户自定义的对象。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度。RDD支持两种操作:transformation操作和action操作。RDD的转化操作是返回一个新的RDD的操作，比如map()和filter()，而action操作则是向驱动器程序返回结果或把结果写入外部系统的操作。比如count()和first()。RDD概述RDD具有如下五个特性：1、一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。2、一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。3、RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。RDD概述4、一个Partitioner，即RDD的分片函数（分区器）。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parentRDDShuffle输出时的分片数量。5、一个列表，存储存取每个Partition的优先位置（preferredlocation）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。感谢大家的聆听RDD创建方式目录/Contents01通过读取文件生成RDD02通过并行化方式创建RDD通过读取文件生成RDD01通过读取文件生成RDDSpark是支持使用任何Hadoop支持的存储系统上的文件创建RDD的，比如说HDFS、Cassandra、HBase以及本地文件。通过调用SparkContext的textFile()方法，可以针对本地文件或HDFS文件创建RDD。1．通过加载本地文件数据创建RDD2．通过加载HDFS文件数据创建RDD通过并行化方式创建RDD02通过并行化方式创建RDD如果要通过并行化集合来创建RDD，需要针对程序中已经存在的集合、数组，调用SparkContext中的parallelize()方法。Spark会将集合中的数据拷贝到集群上去，形成一个分布式的数据集合，也就是一个RDD。即：集合中的部分数据会到一个节点上，而另一部分数据会到其它节点上。然后就可以采用并行的方式来操作这个分布式数据集合。感谢大家的聆听RDD类型操作目录/Contents01

转换算子02行动算子转换算子01转换算子Transformation：转换算子，这类转换并不触发提交作业，完成作业中间过程处理。下面是一些常用的转换算子操作的API。操作介绍map(func)将RDD中的每个元素传入自定义函数，获取一个新的元素，然后用新的元素组成新的RDDfilter(func)对RDD中每个元素进行判断，如果返回true则保留，返回false则剔除flatMap(func)与map类似，但是对每个元素都可以返回一个或多个新元素groupByKey(func)根据key进行分组，每个key对应一个Iterable<value>reduceByKey(func)对每个key对应value进行reduce操作转换算子1.map(func)map(func)操作是对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。2.filter(func)filter(func)操作会筛选出满足条件（即func）的元素，返回一个新的数据集。3.flatMap(func)类似于map，但是每一个输入元素，会被映射为0到多个输出元素(因此，func函数的返回值是一个Seq，而不是单一元素)。在一个由(K,V)对组成的数据集上调用，返回一个(K，Seq[V])对的数据集。4.groupByKey(func)5.reduceByKey(func)顾名思义，reduceByKey就是对元素为KV对的RDD中Key相同的元素的Value进行reduce，因此，Key相同的多个元素的值被reduce为一个值，然后与原RDD中的Key组成一个新的KV对。行动算子02行动算子Action：行动算子，这类算子会触发SparkContext提交Job作业。下面是一些常用的行动算子操作的API。操作介绍

reduce(func)通过函数func聚集数据集中的所有元素。Func函数接受2个参数，返回一个值。这个函数必须是关联性的，确保可以被正确的并发执行collect()在Driver的程序中，以数组的形式，返回数据集的所有元素。count()返回数据集的元素个数take(n)返回一个数组，由数据集的前n个元素组成。first()返回数据集的第一个元素（类似于take（1）)foreach(func)在数据集的每一个元素上，运行函数func。

saveAsTextFile(path)将数据集的元素，以textfile的形式，保存到本地文件系统，hdfs或者任何其它hadoop支持的文件系统。Spark将会调用每个元素的toString方法，并将它转换为文件中的一行文本行动算子1.reduce(func)reduce将RDD中元素两两传递给输入函数，同时产生一个新的值，新产生的值与RDD中下一个元素再被传递给输入函数直到最后只有一个值为止。2.collect()在Driver的程序中，以数组的形式，返回数据集的所有元素。这通常会在使用filter或者其它操作后，返回一个足够小的数据子集再使用，直接将整个RDD集Collect返回，很可能会让Driver程序OOM。3.count()count返回整个RDD的元素个数。我们可以定义一个RDD，使用count()来统计RDD的元素个数。

take和collect操作类似，只是collect操作获取的所有数据，而take操作是获取前n个元素。4.take(n)5.first()first()的作用是返回数据集的第一个元素.我们可以定义一个RDD,使用first()来获取RDD中的第一个元素。6.foreach(func)foreach对RDD中的每个元素都应用func函数操作，不返回RDD和Array，而是返回Uint。感谢大家的聆听RDD之间的依赖关系目录/Contents01

RDD之间的依赖关系RDD之间的依赖关系01RDD之间的依赖关系RDD和它依赖的父RDD的关系有两种不同的类型，即窄依赖（narrowdependency）和宽依赖（widedependency）。RDD之间的依赖关系窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用。总结：窄依赖我们形象的比喻为独生子女。1、窄依赖RDD之间的依赖关系宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition。总结：宽依赖我们形象的比喻为超生。2、宽依赖RDD之间的依赖关系RDD只支持粗粒度转换，即只记录单个块上执行的单个操作。将创建RDD的一系列Lineage（即血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。3、Lineage（血统）感谢大家的聆听RDD机制目录/Contents01持久化机制01容错机制持久化机制01持久化机制多次对某个RDD进行transformation或者action，如果没有做RDD持久化，那么每次都要重新计算一个RDD，会消耗大量时间，降低Spark性能。Spark非常重要的一个功能特性就是可以将RDD持久化在内存中。当对RDD执行持久化操作时，每个节点都会将自己操作的RDD的partition持久化到内存中，并且在之后对该RDD的反复使用中，直接使用内存缓存的partition。这样的话，对于针对一个RDD反复执行多个操作的场景，就只要对RDD计算一次即可，后面直接使用该RDD，而不需要反复计算多次该RDD。持久化机制存储级别说明

MEMORY_ONLY将RDD以反序列化Java对象的形式存储在JVM中。如果内存空间不够，部分分区将不再缓存，在每次需要用到这些数据时重新进行计算。这是默认的存储级别。

MEMORY_AND_DISK将RDD以反序列化Java对象的形式存储在JVM中。如果内存空间不够，将未缓存的数据分区存储到磁盘，在需要使用这些分区时从磁盘读取。

MEMORY_ONLY_SER将RDD以序列化的Java对象的形式进行存储（每个分区为一个字节数组）。这种方式比反序列化的Java对象节省空间，但是在读取时会增加CPU的计算负担。

MEMORY_AND_DISK_SER类似于MEMORY_ONLY_SER，但是溢出的分区会存储到磁盘，而不是在用到它们时重新计算。DISK_ONLY只在磁盘上缓存RDD。

MEMORY_ONLY_2，MEMORY_AND_DISK_2与上面的级别功能相同，只不过每个分区在集群中两个节点上建立副本。需要加上后缀_2,代表的是将每个持久化的数据都复制一份副本，并将副本保存到其他节点上。

OFF_HEAP类似于MEMORY_ONLY_SER，但是将数据存储在off-heapmemory（堆外内存），这需要启动off-heap内存。持久化机制1.如何选择存储级别Spark的存储级别的选择，核心问题是在内存使用率和CPU效率之间进行权衡。2.cache()和persist()的区别cache()和persist()的区别在于，cache()是persist()的一种简化方式，cache()的底层就是调用的persist()的无参版本，同时就是调用persist(MEMORY_ONLY)，将数据持久化到内存中。容错机制02容错机制Spark的计算本质就是对RDD做各种转换，因为RDD是一个不可变只读的集合，因此每次的转换都需要上一次的RDD作为本次转换的输入，因此RDD的lineage描述的是RDD间的相互依赖关系。为了保证RDD中数据的健壮性，RDD数据集通过所谓血统关系（lineage）记住了他是如何其他RDD中演变过来的。Spark将RDD之间的关系规类为宽依赖和窄依赖。Spark会根据Lineage存储的RDD的依赖关系对RDD计算做故障容错，目前Spark的容错策略主要是根据RDD依赖关系重新计算、对RDD做cache、对RDD做checkpoint手段完成RDD计算的故障容错。容错机制SparkRDD实现基于Lineage的容错机制，基于RDD的各项transformation构成了computechain，在部分计算结果丢失的时候可以根据Lineage重新恢复计算。

（1）在窄依赖中，在子RDD的分区丢失，要重算父RDD分区时，父RDD相应分区的所有数据都是子RDD分区的数据，并不存在冗余计算。（2）在宽依赖情况下，丢失一个子RDD分区，重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的，会有一部分数据相当于对应的是未丢失的子RDD分区中需要的数据，这样就会产生冗余计算开销和巨大的性能浪费。感谢大家的聆听SparkSQL简介目录/Contents01什么是SparkSQL02SparkSQL架构什么是SparkSQL01什么是SparkSQLSparkSQL架构与Hive架构相比除了把底层的MapReduce执行引擎更改为Spark还