Ceph资源池pool管理-CFANZ编程社区

概览

每个Spark应用程序都由一个driver program 组成，该驱动程序运行我们编写的main函数，并在集群上执行各种 并行操作。Spark提供的主要抽象是一个 弹性分布式数据集（RDD），它是一个跨集群节点分区的元素集合，可以并行操作。RDD是通过从Hadoop文件系统（或任何其他Hadoop支持的文件系统）中的文件或 driver program 中现有的Scala集合开始并转换它来创建的。用户还可以要求Spark在内存中 持久化RDD，允许它在并行操作中有效地重用。最后，RDD会自动从节点故障中恢复。

Spark中的第二个抽象是可以在并行操作中使用的 共享变量 。默认情况下，当Spark在不同节点上作为一组任务并行运行函数时，它会将函数中使用的每个变量的副本发送给每个任务。有时，一个变量需要在任务之间共享，或者在任务和 driver program 之间共享。Spark支持两种类型的共享变量：广播变量，可用于在所有节点上的内存中缓存一个值，累加器，是仅可“添加”的变量，例如计数器和求和。

一、RDD

弹性分布式数据集(Resilient Distributed Dataset) 简称 RDD，是一个可以并行操作的元素的容错集合。RDD有五大特性：

1、有一系列分区构成（分区即切片）

2、每个函数都作用在每个分区上（并行计算）

3、和其他RDD是有依赖关系的

4、分区类算子只能作用在kv型RDD上

5、Spark会为每个分区计算最佳的位置，并把函数移动过去（计算向数据移动）

RDD支持两种类型的操作：Transformations 和 Actions ，其中Transformations算子有 lazy特性，只要需要将结果返回给驱动程序时，才会触发计算。默认每个转换后的RDD都可以重新计算，也可以使用persist或cache方法将RDD持久化。这样，当后面的算子用这个RDD时就不用重新计算了。

1、Transformations 算子

map(func)

解释：返回一个新的分布式数据集，该数据集是通过函数func传递源的每个元素而形成的

结果：

filter(func)

解释：返回一个新的数据集，该数据集是通过选择func在其上返回true的源的那些元素而形成的。

结果:

flatMap(func)

解释：类似于map，但每个输入项都可以映射到0个或多个输出项（因此func应该返回一个Seq而不是单个项）

结果：

mapPartitions(func)

解释：类似于map，但在RDD的每个分区（块）上单独运行，因此在T类型的RDD上运行时，func必须是Iterator<T>=>Iterator<U>类型。

mapPartitionsWithIndex(func)

解释：类似于map分区，但也为func提供了一个表示分区索引的整数值，因此func必须是类型（Int， Iterator<T>）=>Iterator<U>当在T类型的RDD上运行时。

结果：

sample(withReplacement, fraction, seed)

解释：使用给定的随机数生成器种子对数据的一小部分进行采样，无论是否进行替换。

所谓抽样那么返回的数据占比就没有那么准确

结果：

union(otherDataset)

解释：返回一个新数据集，其中包含源数据集中元素和参数的联合

结果：

intersection(otherDataset)

解释：返回一个新RDD，其中包含源数据集中元素和参数的交集。

结果：

distinct([numTasks]))

解释：返回一个包含源数据集不同元素的新数据集

结果：（无序）

groupByKey([numTasks])

解释：当在（K， V）对的数据集上调用时，返回（K，Iterable<V>）对的数据集

注意：如果您正在分组以执行聚合（例如总和或平均）在每个键上，使用reduceByKey或aggregateByKey会产生更好的结果性能。

注意：默认情况下，输出中的并行级别取决于父RDD的分区数。您可以传递一个可选的numTasks参数来设置不同数量的任务。

结果：

reduceByKey(func, [numTasks])

解释：当在（K， V）对的数据集上调用时，返回（K，V）对的数据集，其中每个键的值使用给定的duce函数func聚合，该函数的类型必须是（V，V）=>V。与groupByKey中一样，可以通过可选的第二个参数配置Reduce任务的数量。

结果：

aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])

解释：当在（K，V）对的数据集上调用时，返回（K，U）对的数据集，其中每个键的值使用给定的组合函数和中性“零”值聚合。允许不同于输入值类型的聚合值类型，同时避免不必要的分配。与groupByKey中一样，可以通过可选的第二个参数来配置Reduce任务的数量。

结果：

sortByKey([ascending], [numTasks])

解释：当在K实现Order的（K， V）对的数据集上调用时，返回（K，V）对的数据集，这些数据集按键按升序或降序排序，如布尔ascending参数中指定的那样。

结果：

join(otherDataset, [numTasks])

解释：当在（K，V）和（K，W）类型的数据集上调用时，返回（K，（V，W））对的数据集，其中每个键的所有元素对。外连接通过leftOuterJoin、rightOuterJoin和fullOuterJoinfullOuterJoin。

结果：

cogroup(otherDataset, [numTasks])

解释：当在（K， V）和（K，W）类型的数据集上调用时，返回（K，（Iterable<V>，Iterable<W>））元组的数据集。此操作也称为groupWith.

结果：

cartesian(otherDataset)

解释：当在T和U类型的数据集上调用时，返回（T，U）对（所有元素对）的数据集。

结果：

pipe(command, [envVars])

解释：通过shell命令（例如Perl或bash脚本）管道RDD的每个分区。RDD元素被写入进程的标准输入，输出到其标准输出的行作为字符串的RDD返回。

因为需要用到脚本，所以必须要到linux环境中

我们执行spark-shell --master yarn 来测试该算子

结果：

coalesce(numPartitions)

解释：将RDD中的分区数量减少到num分区。对于过滤大型数据集后更有效地运行操作很有用。

结果：

repartition(numPartitions)

解释：随机重新洗牌RDD中的数据以创建更多或更少的分区并在它们之间进行平衡。这总是在网络上打乱所有数据。(内部调用的也是coalesce 默认是进行shuffle)

结果：

repartitionAndSortWithinPartitions(partitioner)

解释：根据给定的分区器重新分区RDD，并且在每个结果分区中，按键对记录进行排序。这比调用repartition然后在其中排序更有效每个分区，因为它可以将排序向下推到洗牌机械中。

repartitionAndSortWithinPartitions是Spark官网推荐的一个算子，官方建议，如果需要在repartition重分区之后，还要进行排序，建议直接使用repartitionAndSortWithinPartitions算子。因为该算子可以一边进行重分区的shuffle操作，一边进行排序。shuffle与sort两个操作同时进行，比先shuffle再sort来说，性能可能是要高的。

结果：

2、Actions 算子

reduce(func)

解释：使用函数func（它接受两个参数并返回一个）聚合数据集的元素。该函数应该是可交换和关联的，以便可以并行正确计算。

结果：

collect()

解释：在驱动程序中以数组的形式返回数据集的所有元素。这通常在返回足够小的数据子集的过滤器或其他操作之后很有用。

结果：

count()

解释：返回数据集中元素的数量。

结果：

first()

解释：返回数据集的第一个元素（类似于take(1) ）。

结果：

take(n)

解释：返回包含数据集前n个元素的数组

结果：

takeSample(withReplacement, num, [seed])

解释：返回一个数组，其中包含数据集num元素的随机样本，有或没有替换，可选地预先指定随机数生成器种子。

结果：

takeOrdered(n, [ordering])

解释：使用它们的自然顺序或自定义比较器返回RDD的前n个元素

结果：

saveAsTextFile(path)

解释：将数据集的元素作为文本文件（或一组文本文件）写入本地文件系统、HDFS或任何其他Hadoop支持的文件的给定目录中系统。Spark将在每个元素上调用toString将其转换为文件中的一行文本。

saveAsSequenceFile(path)

解释：将数据集的元素写入本地文件系统、HDFS或任何其他Hadoop支持的文件系统的给定路径中的Hadoop Sequence File。这在实现Hadoop的Writable接口的键值对的RDD上可用。在Scala中，它也适用于隐式转换为Writable的类型（Spark包括Int、Double、String等基本类型的转换）。

saveAsObjectFile(path)

解释：使用Java序列化以简单格式写入数据集的元素，然后可以使用 SparkContext.objectFile()。

countByKey()

解释：仅在（K，V）类型的RDD上可用。返回（K，Int）对的哈希图以及每个键的计数。

结果：

foreach(func)（上面已经演示过）

解释：对数据集的每个元素运行一个函数func。这通常用于副作用，例如更新累加器或与外部存储系统交互。

注意：在foreach()之外修改除Accumulator之外的变量可能会导致未定义的行为。有关详细信息，请参阅了解闭包。

3、Shuffle操作

Spark中的某些操作会触发称为shuffle的事件。shuffle是Spark重新分配数据的机制，以便在分区之间以不同的方式分组。这通常涉及跨 executors 和节点复制数据，因此shuffle是复杂且成本高昂的操作。

我们以 reduceByKey 为例说明：

reduceByKey 生成一个新的RDD，相同key的所有value会被组合到一起，这就需要将本分区中不同的key发送到不同的节点。这对于Spark来说是一个全向操作。它必须从所有分区中读取以查找所有key的所有value，然后将跨分区的value聚集在一起以计算每个key的最终结果- 这就是Shuffle。

尽管新洗牌数据的每个分区中的元素集是确定的，分区本身的顺序也是确定的，但这些元素的顺序不是。如果希望在洗牌后可预测地排序数据，那么可以使用：

mapPartitions（对每个分区进行排序，例如，使用.sorted）

repartitionAndSortWithinPartitions（在重新分区的同时有效地对分区进行排序）

sortBy（创建全局排序的RDD）

由于Shuffle涉及磁盘I/O、数据序列化和网络输入输出。为了整理这些数据，Spark会产生map任务和reduce任务来实现（这里说的并不是MapReduce中的mapTask和reduceTask）

在内部，单个map任务的结果会保存在内存中，当数据不合适再内存中时会溢写到磁盘（会产生磁盘IO开销和垃圾回收机制）

Shuffle会在磁盘产生大量的中间文件，这些文件会一直保留，直到不再使用相关RDD并被垃圾收集。这是为了再次使用RDD时提升Shuffle的性能。

4、持久化

当我们有迭代算法的需求时，我们就需要将之前的RDD进行缓存或者持久化，可以直接使用persist()或cache()来实现。Spark的缓存也是容错的，如果发现缓存的RDD有分区丢失，会通过血缘关系再次创建它。

每个RDD都可以选择不同的持久化级别，比如磁盘、内存。这些级别通过传递一个 StorageLevel对象来实现。

存储级别	含义
MEMORY_ONLY	将RDD存储为JVM中反序列化的Java对象。如果RDD不适合内存，某些分区将不会被缓存，并且每次需要时都会立即重新计算。这是默认级别。
MEMORY_AND_DISK	将RDD存储为JVM中反序列化的Java对象。如果RDD不适合内存，则将不适合磁盘的分区存储在磁盘上，并在需要时从那里读取它们。
MEMORY_ONLY_SER	只适用于Java和Scala，将RDD存储为序列化的Java对象（每个分区一个字节数组）。这通常比反序列化对象更节省空间，尤其是在使用快速序列化程序，但读取更占用CPU。
MEMORY_AND_DISK_SER	只适用于Java和Scala，类似于MEMORY_ONLY_SER，但是将不适合内存的分区溢出到磁盘，而不是每次需要时立即重新计算它们。
DISK_ONLY	仅将RDD分区存储在磁盘上
MEMORY_ONLY_2、MEMORY_AND_DISK_2等。	与上面的级别相同，但在两个集群节点上复制每个分区。
OFF_HEAP	（实验阶段）类似于MEMORY_ONLY_SER，但将数据存储在堆外内存。这需要启用堆外内存。