Spark_04_弹性分布式数据集介绍

Spark 围绕弹性分布式数据集（RDD）的概念展开，它是一组可以并行操作的容错元素集合。

创建 RDD 有两种方式：在驱动程序中并行化现有的集合，或者引用外部存储系统中的数据集，例如共享文件系统、HDFS、HBase 或任何提供 Hadoop InputFormat 的数据源。

Resilient Distributed Datasets

1. RDD 的核心特性

• 分布式(Distributed)：RDD区别于传统的数据集（集合），最大的特点就是它是分布式的，也就是说，它的数据会被切割成多个分区（partition），每个分区可以分布在集群中的不同节点上进行运算。 这样做的好处是可以并行的大规模处理数据，第二个好处就是方便编程。

• 弹性（Resilient）：容错性。如果某个rdd转换的过程中，部分分区数据丢失，spark可以通过rdd的关系（血统lineage）信息重新计算丢失的分区，而不需要重新计算整个数据集。

• 不可变性（immutable）：RDD是不可变的，一旦被创建之后，就不能再修改了，所有的转换操作都会生成一个新的RDD，而不是修改原始的RDD。

• 类型化（Typed）：RDD和集合一样，都是强类型的，可以存储任何类型的数据，但是一个RDD中的数据类型必须都相同。

2. RDD 的创建方式

1. 通过并行化接口从集合创建RDD （常用于测试）
2. 通过外部数据创建（本地文件/hdfs文件）

3. RDD 的操作

RDD支持两种操作方式：transformation和action

RDD的转换操作是惰性求值的，意思是所有的转换操作，在代码执行过程中，不会立即执行，而是记录下操作逻辑，直到遇到了action才会触发计算。

因为如果没有action，说明这个计算结果没有人使用，那么就不计算了。

1. Transformation（转换操作）

Transformation	Meaning
map(func)	Return a new distributed dataset formed by passing each element of the source through a function func. 返回由通过函数 func 对源数据集中的每个元素进行传递而形成的新分布式数据集。
filter(func)	Return a new dataset formed by selecting those elements of the source on which func returns true.
flatMap(func)	Similar to map, but each input item can be mapped to 0 or more output items (so func should return a Seq rather than a single item).
mapPartitions(func)	Similar to map, but runs separately on each partition (block) of the RDD, so func must be of type Iterator => Iterator when running on an RDD of type T.
groupByKey([numPartitions])	When called on a dataset of (K, V) pairs, returns a dataset of (K, Iterable) pairs. Note: If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using reduceByKey or aggregateByKey will yield much better performance. Note: By default, the level of parallelism in the output depends on the number of partitions of the parent RDD. You can pass an optional numPartitions argument to set a different number of tasks.
reduceByKey(func, [numPartitions])	When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function func, which must be of type (V,V) => V. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument.
aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions])	When called on a dataset of (K, V) pairs, returns a dataset of (K, U) pairs where the values for each key are aggregated using the given combine functions and a neutral “zero” value. Allows an aggregated value type that is different than the input value type, while avoiding unnecessary allocations. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument.
repartition(numPartitions)	Reshuffle the data in the RDD randomly to create either more or fewer partitions and balance it across them. This always shuffles all data over the network.
coalesce(numPartitions)	Decrease the number of partitions in the RDD to numPartitions. Useful for running operations more efficiently after filtering down a large dataset.

2. action

   action会触发真正的计算，并将结果返回到 Driver或者保存到外部进行存储。

reduce(func)	Aggregate the elements of the dataset using a function func (which takes two arguments and returns one). The function should be commutative and associative so that it can be computed correctly in parallel.
collect()	Return all the elements of the dataset as an array at the driver program. This is usually useful after a filter or other operation that returns a sufficiently small subset of the data.
count()	Return the number of elements in the dataset.
first()	Return the first element of the dataset (similar to take(1)).
take(n)	Return an array with the first n elements of the dataset.
takeSample(withReplacement, num, [seed])	Return an array with a random sample of num elements of the dataset, with or without replacement, optionally pre-specifying a random number generator seed.
takeOrdered(n, [ordering])	Return the first n elements of the RDD using either their natural order or a custom comparator.
saveAsTextFile(path)	Write the elements of the dataset as a text file (or set of text files) in a given directory in the local filesystem, HDFS or any other Hadoop-supported file system. Spark will call toString on each element to convert it to a line of text in the file.
saveAsSequenceFile(path) (Java and Scala)	Write the elements of the dataset as a Hadoop SequenceFile in a given path in the local filesystem, HDFS or any other Hadoop-supported file system. This is available on RDDs of key-value pairs that implement Hadoop’s Writable interface. In Scala, it is also available on types that are implicitly convertible to Writable (Spark includes conversions for basic types like Int, Double, String, etc).
saveAsObjectFile(path) (Java and Scala)	Write the elements of the dataset in a simple format using Java serialization, which can then be loaded using SparkContext.objectFile().
countByKey()	Only available on RDDs of type (K, V). Returns a hashmap of (K, Int) pairs with the count of each key.
foreach(func)	Run a function func on each element of the dataset. This is usually done for side effects such as updating an Accumulator or interacting with external storage systems. Note: modifying variables other than Accumulators outside of the foreach() may result in undefined behavior. See Understanding closures for more details.

4. 依赖关系

宽窄依赖 –> stage划分

• 宽依赖（Wide Dependency）

  每个父RDD的分区可能被多个子RDD的分区使用，这种就叫做宽依赖。

  通常会产生宽依赖的算子包含： groupByKey、reduceByKey、repartition、distinct

• 窄依赖（Narrow Dependency）

每个父RDD的分区最多被一个子RDD的分区使用，这种就叫做窄依赖。

通常会产生窄依赖的算子包含： map、filter、mapPartition、sample、union

• 宽依赖必然会有shuffle过程，shuffle的本质是数据的跨节点计算，因此在划分stage的时候，遇到了shuffle（宽依赖）就会切割stage（切割血缘）

5. 分区和并行度

在spark中，所谓的分区（partition）就是指的是数据分布的物理单元。分区的数量会影响任务的数量，分区越多，task越多

并行度（parallelism）是任务执行的并发能力，并行度指的是同一时间内，有多少个task参与计算，并行度越高，通常任务的性能越好。

• 分区

  • 读取文件时，分区数按照文件的快大小分隔分区（hdfs）
  • 并行化集合时：由参数numSlices指定分区数（不指定的情况下默认为CPU核心数）
  • shuffle后： spark.sql.shuffle.partitions 默认200，可以在spark sql客户端中通过set来设置
  • 自定义分区器：按照 numPartitions来控制分区数

• 并行度

  • 并行度取决于 executor数量（人） * 每个executor的CPU核心数（每个人同一时间能做几件事情）

合理的调整分区数大小和并行度大小，可以达到优化任务执行性能的效果

分区数不是越多越好，太多的话每个task操作的数据量很小，并且task需要多轮才能执行完成，太少的话，部分executor没活干（无法充分利用集群资源）

一般情况下，我们尽量让分区数接近spark集群运行时的可用核心数，避免资源限制或任务过载。

通常分区数为总并行度的1~4倍较为合理，当然也要考虑每个分区处理的数据量。

cache或者persist接口只有在遇到了action之后，才会触发真正的执行。

6. 持久化和缓存

持久化：就是把数据存储到一个地方（磁盘），需要用的时候再拿出来

缓存：就是把数据存储到一个地方（内存），需要用的时候再拿出来

通常持久化就是保留到永久存储介质，通常为磁盘；缓存就是保存到内存中。但是在spark中，不管是内存还是磁盘，在任务结束的时候都会销毁，所以spark中的持久化和缓存其实是相同的概念，只不过cache是特殊的persist

常用的是 MEMORY_ONLY（消耗内存多，但是块）、MEMORY_AND_DISK

cache本质就是 StorageLevel.MEMORY_ONLY 的persist

torage Level	Meaning
MEMORY_ONLY	Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, some partitions will not be cached and will be recomputed on the fly each time they’re needed. This is the default level.
MEMORY_AND_DISK	Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, store the partitions that don’t fit on disk, and read them from there when they’re needed.
MEMORY_ONLY_SER (Java and Scala)	Store RDD as serialized Java objects (one byte array per partition). This is generally more space-efficient than deserialized objects, especially when using a fast serializer, but more CPU-intensive to read.
MEMORY_AND_DISK_SER (Java and Scala)	Similar to MEMORY_ONLY_SER, but spill partitions that don’t fit in memory to disk instead of recomputing them on the fly each time they’re needed.
DISK_ONLY	Store the RDD partitions only on disk.
MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc.	Same as the levels above, but replicate each partition on two cluster nodes.
OFF_HEAP (experimental)	Similar to MEMORY_ONLY_SER, but store the data in off-heap memory. This requires off-heap memory to be enabled.

需要注意，这里的持久化不是真的持久化，这个持久化只在spark application的生命周期中有效，一旦application结束，persist也会被清理。

7. checkpoint

很少用，通常是在SparkStreaming中使用，用于容错。

设置检查点（checkpoint）方式，本质上是将RDD写入磁盘进行存储。当RDD在进行宽依赖运算时，只需要在中间阶段设置一个检查点进行容错，即通过 Spark中的sparkContext对象调用setCheckpoint()方法，设置一个容错文件系统目录(如 HDFS)作为检查点checkpoint，将checkpoint 的数据写入之前设置的容错文件系统中进行高可用的持久化存储， 若是后面有节点出现宕机导致分区数据丢失,则可以从作为检查点的RDD开始重新计算，不需要进行从头到尾的计 算，这样就会减少开销。

简单理解，Checkpoint 就是一种将 RDD 或 DataFrame/Dataset 持久化到可靠存储（如 HDFS 或本地文件系统）的机制，用于切断 RDD 的血缘关系（Lineage），避免任务失败时从头重新计算。

checkpoint还有一个作用，可以用于中断任务后，重启任务加载上个任务的数据