spark系列-2、Spark 核心数据结构：弹性分布式数据集 RDD

一、RDD(弹性分布式数据集)

RDD 是 Spark 最核心的数据结构，RDD（Resilient Distributed Dataset）全称为弹性分布式数据集，是 Spark 对数据的核心抽象，也是最关键的抽象，它实质上是一组分布式的 JVM 不可变对象集合，不可变决定了它是只读的，所以 RDD 在经过变换产生新的 RDD 时，原有 RDD 不会改变。

1.1、设计背景

在实际应用中，存在许多迭代式计算，这些应用场景的共同之处是，不同计算阶段之间会重用中间结果，即一个阶段的输出结果会作为下一个阶段的输入。但是，目前的MapReduce框架都是把中间结果写入到HDFS中，带来了大量的数据复制、磁盘IO和序列化开销。显然，如果能将结果保存在内存当中，就可以大量减少IO。RDD就是为了满足这种需求而出现的，它提供了一个抽象的数据架构，我们不必担心底层数据的分布式特性，只需将具体的应用逻辑表达为一系列转换处理，不同RDD之间的转换操作形成依赖关系，可以实现管道化，从而避免了中间结果的落地存储，大大降低了数据复制、磁盘IO和序列化开销。

1.2、RDD概念

一个RDD就是一个分布式对象集合，本质上是一个只读的分区记录集合，每个RDD可以分成多个分区，每个分区就是一个数据集片段（HDFS上的块），并且一个RDD的不同分区可以被保存到集群中不同的节点上，从而可以在集群中的不同节点上进行并行计算。RDD提供了一种高度受限的共享内存模型，即RDD是只读的记录分区的集合，不能直接修改，只能基于稳定的物理存储中的数据集来创建RDD，或者通过在其他RDD上执行确定的转换操作（如map、join和groupBy）而创建得到新的RDD。RDD提供了一组丰富的操作以支持常见的数据运算，分为“行动”（Action）和“转换”（Transformation）两种类型，前者用于执行计算并指定输出的形式，后者指定RDD之间的相互依赖关系。两类操作的主要区别是，转换操作（比如map、filter、groupBy、join等）接受RDD并返回RDD，而行动操作（比如count、collect等）接受RDD但是返回非RDD（即输出一个值或结果）。

RDD典型的执行过程如下：

• RDD读入外部数据源（或者内存中的集合）进行创建；
• RDD经过一系列的“转换”操作，每一次都会产生不同的RDD，供给下一个“转换”使用；
• 最后一个RDD经“行动”操作进行处理，并输出到外部数据源（或者变成Scala/JAVA集合或变量）。
• 需要说明的是，RDD采用了惰性调用，即在RDD的执行过程中，真正的计算发生在RDD的“行动”操作，对于“行动”之前的所有“转换”操作，Spark只是记录下“转换”操作应用的一些基础数据集以及RDD生成的轨迹，即相互之间的依赖关系，而不会触发真正的计算（Action才会计算）。
• 

如下图所示：从输入中逻辑上生成A和C两个RDD，经过一系列“转换”操作，逻辑上生成了F（也是一个RDD），之所以说是逻辑上，是因为这时候计算并没有发生，Spark只是记录了RDD之间的生成和依赖关系。当F要进行输出时，也就是当F进行“行动”操作的时候，Spark才会根据RDD的依赖关系生成DAG，并从起点开始真正的计算。

这一系列处理称为一个“血缘关系（Lineage）”，即DAG拓扑排序的结果。采用惰性调用，通过血缘关系连接起来的一系列RDD操作就可以实现管道化（pipeline）（A->B通过内存传数据，避免了IO），避免了多次转换操作之间数据同步的等待，而且不用担心有过多的中间数据，因为这些具有血缘关系的操作都管道化了，一个操作得到的结果不需要保存为中间数据，而是直接管道式地流入到下一个操作进行处理。同时，这种通过血缘关系就是把一系列操作进行管道化连接的设计方式，也使得管道中每次操作的计算变得相对简单，保证了每个操作在处理逻辑上的单一性；相反，在MapReduce的设计中，为了尽可能地减少MapReduce过程，在单个MapReduce中会写入过多复杂的逻辑。

 程序示例：

val conf = new SparkConf
val sparkContext = new SparkContext(conf)
val sourceRDD :RDD = sparkContext.textFile(logFile)
val countRDD = sourceRDD.filter(_.contains("hello world")).cache().count()
println(countRDD)

可以看出，一个Spark应用程序，基本是基于RDD的一系列计算操作。

• 第2行代码用于创建JavaSparkContext对象；
• 第3行代码从HDFS文件中读取数据创建一个RDD；
• 第4行代码对sourceRDD进行转换操作得到一个新的RDD，即countRDD；
• sourceRDD.cache()表示对lines进行持久化，把它保存在内存或磁盘中（这里采用cache接口把数据集保存在内存中），方便后续重复使用，当数据被反复访问时（比如查询一些热点数据，或者运行迭代算法），这是非常有用的，而且通过cache()可以缓存非常大的数据集，支持跨越几十甚至上百个节点（每个机器存自己负责的那一部分 Executor）；sourceRDD.count()是一个行动操作，用于计算一个RDD集合中包含的元素个数。

这个程序的执行过程如下：

• 创建这个Spark程序的执行上下文，即创建SparkContext对象；
• 从外部数据源（即HDFS文件）中读取数据创建sourceRDD对象；
• 构建起sourceRDD和countRDD之间的依赖关系，形成DAG图，这时候并没有发生真正的计算，只是记录转换的轨迹；
• 执行action代码时，count()是一个行动类型的操作，触发真正的计算，并把结果持久化到内存中，最后计算出sourceRDD中包含的元素个数。

1.3、RDD特性

总体而言，Spark采用RDD以后能够实现高效计算的主要原因如下：

• 高效的容错性。现有的分布式共享内存、键值存储、内存数据库等，为了实现容错，必须在集群节点之间进行数据复制或者记录日志，也就是在节点之间会发生大量的数据传输，这对于数据密集型应用而言会带来很大的开销。在RDD的设计中，数据只读，不可修改，如果需要修改数据，必须从父RDD转换到子RDD，由此在不同RDD之间建立了血缘关系。所以，RDD是一种天生具有容错机制的特殊集合，不需要通过数据冗余的方式（比如详细的记录操作的日志）实现容错，而只需通过RDD父子依赖（血缘）关系重新计算得到丢失的分区来实现容错，无需回滚整个系统，这样就避免了数据复制的高开销，而且重算过程可以在不同节点之间并行进行，实现了高效的容错。此外，RDD提供的转换操作都是一些粗粒度的操作（比如map、filter和join），RDD依赖关系只需要记录这种粗粒度的转换操作，而不需要记录具体的数据和各种细粒度操作的日志（比如对哪个数据项进行了修改），这就大大降低了数据密集型应用中的容错开销
• 中间结果持久化到内存。数据在内存中的多个RDD操作之间进行传递，不需要“落地”到磁盘上，避免了不必要的读写磁盘开销；
• 存放的数据可以是Java对象，避免了不必要的对象序列化和反序列化开销。

1.4、RDD之间的依赖关系

RDD中不同的操作会使得不同RDD中的分区会产生不同的依赖。RDD中的依赖关系分为窄依赖（Narrow Dependency）与宽依赖（伴随shuffle）（Wide Dependency）
两种依赖之间的区别：

• 窄依赖表现为一个或多个父RDD的分区对应于一个子RDD的分区；
• 宽依赖则表现为存在一个父RDD的一个分区对应一个子RDD的多个分区

总体而言，如果父RDD的一个分区只被一个子RDD的一个分区所使用就是窄依赖，否则就是宽依赖。窄依赖典型的操作包括map、filter、union等，宽依赖典型的操作包括groupByKey，sortByKey等。对于连接（join）操作，可以分为两种情况。

1. 对输入进行协同划分，属于窄依赖。所谓协同划分（co-partitioned）是指多个父RDD的某一分区的所有“键（key）”，落在子RDD的同一个分区内，不会产生同一个父RDD的某一分区，落在子RDD的两个分区的情况。
2. 对输入做非协同划分，属于宽依赖。

对于窄依赖的RDD，可以以流水线的方式计算所有父分区，不会造成网络之间的数据混合。对于宽依赖的RDD，则通常伴随着Shuffle操作，即首先需要计算好所有父分区数据，然后在节点之间进行Shuffle。

Spark的这种依赖关系设计，使其具有了天生的容错性，大大加快了Spark的执行速度。因为，RDD数据集通过“血缘关系”记住了它是如何从其它RDD中演变过来的，血缘关系记录的是粗颗粒度的转换操作行为，当这个RDD的部分分区数据丢失时，它可以通过血缘关系获取足够的信息来重新运算和恢复丢失的数据分区，由此带来了性能的提升。相对而言，在两种依赖关系中，窄依赖的失败恢复更为高效，它只需要根据父RDD分区重新计算丢失的分区即可（不需要重新计算所有分区），而且可以并行地在不同节点进行重新计算。而对于宽依赖而言，单个节点失效通常意味着重新计算过程会涉及多个父RDD分区，开销较大。此外，Spark还提供了数据检查点和记录日志，用于持久化中间RDD，从而使得在进行失败恢复时不需要追溯到最开始的阶段。在进行故障恢复时，Spark会对数据检查点开销和重新计算RDD分区的开销进行比较，从而自动选择最优的恢复策略。

1.5、阶段的划分

Spark通过分析各个RDD的依赖关系生成了DAG，再通过分析各个RDD中的分区之间的依赖关系来决定如何划分阶段。
具体划分方法是：
在DAG中进行反向解析，遇到宽依赖就断开，遇到窄依赖就把当前的RDD加入到当前的阶段中；
将窄依赖尽量划分在同一个阶段中，可以实现流水线计算。例如，假设从HDFS中读入数据生成3个不同的RDD（即A、C和E），通过一系列转换操作后再将计算结果保存回HDFS。对DAG进行解析时，在依赖图中进行反向解析，由于从RDD A到RDD B的转换以及从RDD B和F到RDD G的转换，都属于宽依赖，因此，在宽依赖处断开后可以得到三个阶段，即阶段1、阶段2和阶段3。可以看出，在阶段2中，从map到union都是窄依赖，这两步操作可以形成一个流水线操作，比如，分区7通过map操作生成的分区9，可以不用等待分区8到分区10这个转换操作的计算结束，而是继续进行union操作，转换得到分区13，这样流水线执行大大提高了计算的效率。

由上述论述可知，把一个DAG图划分成多个“阶段”以后，每个阶段都代表了一组关联的、相互之间没有Shuffle依赖关系的任务组成的任务集合。每个任务集合会被提交给任务调度器（TaskScheduler）进行处理，由任务调度器将任务分发给Executor运行。

1.6、RDD在Spark架构中的运行过程

（1）创建RDD对象；
（2）SparkContext负责计算RDD之间的依赖关系，构建DAG；
（3）DAGScheduler负责把DAG图分解成多个阶段，每个阶段中包含了多个任务，每个任务会被任务调度器分发给各个工作节点（Worker Node）上的Executor去执行。

 1.7、创建RDD

• 并行化集合
  • 这种 RDD 纯粹是为了学习，将内存中的集合变量转换为 RDD，没太大实际意义。

  • val conf: SparkConf = new SparkConf()
        conf.setMaster("local[*]")
          .setAppName("wordCount")
        val context: SparkContext = new SparkContext(conf)
    
        val sourceRDD: RDD[String] = context.parallelize(
          List(
            "a b c d e",
            "a b c d",
            "a b c",
            "a b",
            "a")
        )

• 从 HDFS 中读取，这种生成 RDD 的方式是非常常用的

  • val conf: SparkConf = new SparkConf()
        conf.setMaster("local[*]")
          .setAppName("wordCount")
        val context: SparkContext = new SparkContext(conf)
        context.textFile("hdfs://namenode:8020/user/me/wiki.txt")

• 从外部数据源读取
  • Spark 从 MySQL 中读取数据返回的 RDD 类型是 JdbcRDD，顾名思义，是基于 JDBC 读取数据的，这点与 Sqoop 是相似的，但不同的是 JdbcRDD 必须手动指定数据的上下界，也就是以 MySQL 表某一列的最值作为切分分区的依据。

  • val conf: SparkConf = new SparkConf()
        conf.setMaster("local[*]")
          .setAppName("wordCount")
        val context: SparkContext = new SparkContext(conf)
        val lowerBound = 1
        val upperBound = 1000
        val numPartition = 10
        val rdd = new JdbcRDD(context,() => {
          Class.forName("com.mysql.jdbc.Driver").newInstance()
          DriverManager.getConnection("jdbc:mysql://localhost:3306/db", "root", "123456")
        },
          "SELECT content FROM mysqltable WHERE ID >= ? AND ID <= ?",
          lowerBound,
          upperBound,
          numPartition,
          r => r.getString(1)
        )

  • 既然是基于 JDBC 进行读取，那么所有支持 JDBC 的数据库都可以通过这种方式进行读取，也包括支持 JDBC 的分布式数据库，但是需要注意的是，从代码可以看出，这种方式的原理是利用多个 Executor 同时查询互不交叉的数据范围，从而达到并行抽取的目的。但是这种方式的抽取性能受限于 MySQL 的并发读性能，单纯提高 Executor 的数量到某一阈值后，再提升对性能影响不大。
  • 上面的是通过 JDBC 读取数据库的方式，对于 HBase 这种分布式数据库来说，情况有些不同，HBase 这种分布式数据库，在数据存储时也采用了分区的思想，HBase 的分区名为 Region，那么基于 Region 进行导入这种方式的性能就会比上面那种方式快很多，是真正的并行导入。

  • //val spark: SparkSession = .......
    val sc = spark.sparkcontext
    val tablename = "your_hbasetable"  
    val conf = HBaseConfiguration.create()  
    conf.set("hbase.zookeeper.quorum", "zk1,zk2,zk3")  
    conf.set("hbase.zookeeper.property.clientPort", "2181")  
    conf.set(TableInputFormat.INPUT_TABLE, tablename)  
    val rdd= sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],  
    classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],  
    classOf[org.apache.hadoop.hbase.client.Result]) 
    // 利用HBase API解析出行键与列值
    rdd_three.foreach{case (_,result) => {    
        val rowkey = Bytes.toString(result.getRow)  
        val value1 = Bytes.toString(result.getValue("cf".getBytes,"c1".getBytes))
    }

  • HBase 有一个第三方组件叫 Phoenix，可以让 HBase 支持 SQL 和 JDBC，在这个组件的配合下，第一种方式也可以用来抽取 HBase 的数据，此外，Spark 也可以读取 HBase 的底层文件 HFile，从而直接绕过 HBase 读取数据。
  • 通过第三方库的支持，Spark 几乎能够读取所有的数据源，例如 Elasticsearch。

 1.8、PairRDD

PairRDD 与其他 RDD 并无不同，只不过它的数据类型是 Tuple2[K,V]，表示键值对，因此这种 RDD 也被称为 PairRDD，泛型为 RDD[(K,V)]，而普通 RDD 的数据类型为 Int、String 等。这种数据结构决定了 PairRDD 可以使用某些基于键的算子，如分组、汇总等。PairRDD 可以由普通 RDD 转换得到：

//val spark: SparkSession = .......
val a = spark.sparkcontext.textFile("/user/me/wiki").map(x => (x,x))

二、Transformations算子

下面列出了Spark常用的transformation操作。详细的细节请参考RDD API文档(Scala、Java、Python、R)和键值对RDD方法文档(Scala、Java)。

• map(func)
  • 将原来RDD的每个数据项，使用map中用户自定义的函数func进行映射，转变为一个新的元素，并返回一个新的RDD。
• filter(func)
  • 使用函数func对原RDD中数据项进行过滤，将符合func中条件的数据项组成新的RDD返回。
• flatMap(func)
  • 类似于map，但是输入数据项可以被映射到0个或多个输出数据集合中，所以函数func的返回值是一个数据项集合而不是一个单一的数据项。
• mapPartitions(func)
  • 类似于map，但是该操作是在每个分区上分别执行，所以当操作一个类型为T的RDD时func的格式必须是Iterator<T> => Iterator<U>。即mapPartitions需要获取到每个分区的迭代器，在函数中通过这个分区的迭代器对整个分区的元素进行操作。
• mapPartitionsWithIndex(func)
  • 类似于mapPartitions，但是需要提供给func一个整型值，这个整型值是分区的索引，所以当处理T类型的RDD时，func的格式必须为(Int, Iterator<T>) => Iterator<U>。
• union(otherDataset)
  • 返回原数据集和参数指定的数据集合并后的数据集。使用union函数时需要保证两个RDD元素的数据类型相同，返回的RDD数据类型和被合并的RDD元素数据类型相同。该操作不进行去重操作，返回的结果会保存所有元素。如果想去重，可以使用distinct（）。
• intersection(otherDataset)
  • 返回两个数据集的交集。
• distinct([numTasks]))
  • 将RDD中的元素进行去重操作。
• groupByKey([numTasks])
  • 操作（K,V）格式的数据集，返回 (K, Iterable)格式的数据集。
  • 注意，如果分组是为了按key进行聚合操作（例如，计算sum、average)，此时使用reduceByKey或aggregateByKey计算效率会更高。
  • 注意，默认情况下，并行情况取决于父RDD的分区数，但可以通过参数numTasks来设置任务数。
• reduceByKey(func, [numTasks])
  • 使用给定的func，将（K,V）对格式的数据集中key相同的值进行聚集，其中func的格式必须为(V,V) => V。可选参数numTasks可以指定reduce任务的数目。
• aggregateByKey(zeroValue)(seqOp, combOp,[numTasks])
  • 对（K,V）格式的数据按key进行聚合操作，聚合时使用给定的合并函数和一个初试值，返回一个(K,U)对格式数据。需要指定的三个参数:zeroValue为在每个分区中,对key值第一次读取V类型的值时,使用的U类型的初始变量；seqOp用于在每个分区中，相同的key中V类型的值合并到zeroValue创建的U类型的变量中。combOp是对重新分区后两个分区中传入的U类型数据的合并函数。
• sortByKey([ascending], [numTasks])
  • (K,V)格式的数据集，其中K已实现了Ordered，经过sortByKey操作返回排序后的数据集。指定布尔值参数ascending来指定升序或降序排列。

• join(otherDataset, [numTasks])
  • 用于操作两个键值对格式的数据集，操作两个数据集(K,V)和(K,W)返回(K, (V, W))格式的数据集。通过leftOuterJoin、rightOuterJoin、fullOuterJoin完成外连接操作。

• cogroup(otherDataset, [numTasks])
  • 用于操作两个键值对格式数据集(K,V)和(K,W)，返回数据集格式为 (K,(Iterable, Iterable)) 。这个操作也称为groupWith。对在两个RDD中的Key-Value类型的元素，每个RDD相同Key的元素分别聚合为一个集合，并且返回两个RDD中对应Key的元素集合的迭代器。

• cartesian(otherDataset)
  • 对类型为T和U的两个数据集进行操作，返回包含两个数据集所有元素对的（T,U）格式的数据集。即对两个RDD内的所有元素进行笛卡尔积操作。

• pipe(command, [envVars])
  • 以管道(pipe)方式将 RDD的各个分区(partition)使用 shell命令处理（比如一个 Perl或 bash脚本）。 RDD的元素会被写入进程的标准输入(stdin)，将进程返回的一个字符串型 RDD(RDD of strings)，以一行文本的形式写入进程的标准输出(stdout)中。

• coalesce(numPartitions)
  • 把RDD的分区数降低到通过参数numPartitions指定的值。在得到的更大一些数据集上执行操作，会更加高效。

• repartition(numPartitions)
  • 随机地对RDD的数据重新洗牌(Reshuffle)，从而创建更多或更少的分区，以平衡数据。总是对网络上的所有数据进行洗牌(shuffles)。
• repartitionAndSortWithinPartitions(partitioner)
  • 根据给定的分区器对RDD进行重新分区，在每个结果分区中，按照key值对记录排序。这在每个分区中比先调用repartition再排序效率更高，因为它可以将排序过程在shuffle操作的机器上进行。

3、Actions算子

下面列出了Spark支持的常用的action操作。详细请参考RDD API文档(Scala、Java、Python、R)和键值对RDD方法文档(Scala、Java)。

• reduce(func)
  • 使用函数func聚集数据集中的元素，这个函数func输入为两个元素，返回为一个元素。这个函数应该符合结合律和交换了，这样才能保证数据集中各个元素计算的正确性。

• collect()
  • 在驱动程序中，以数组的形式返回数据集的所有元素。通常用于filter或其它产生了大量小数据集的情况。

• count()
  • 返回数据集中元素的个数。

• first()
  • 返回数据集中的第一个元素（类似于take(1)）。

• take(n)
  • 返回数据集中的前n个元素。

• takeOrdered(n, [ordering])
  • 返回RDD按自然顺序或自定义顺序排序后的前n个元素。

• saveAsTextFile(path)
  • 将数据集中的元素以文本文件（或文本文件集合）的形式保存到指定的本地文件系统、HDFS或其它Hadoop支持的文件系统中。Spark将在每个元素上调用toString方法，将数据元素转换为文本文件中的一行记录。

• saveAsSequenceFile(path) (Java and Scala)
  • 将数据集中的元素以Hadoop Sequence文件的形式保存到指定的本地文件系统、HDFS或其它Hadoop支持的文件系统中。该操作只支持对实现了Hadoop的Writable接口的键值对RDD进行操作。在Scala中，还支持隐式转换为Writable的类型（Spark包括了基本类型的转换，例如Int、Double、String等等)。

• saveAsObjectFile(path) (Java and Scala)
  • 将数据集中的元素以简单的Java序列化的格式写入指定的路径。这些保存该数据的文件，可以使用SparkContext.objectFile()进行加载。

• countByKey()
  • 仅支持对（K,V）格式的键值对类型的RDD进行操作。返回（K,Int）格式的Hashmap，(K,Int)为每个key值对应的记录数目。

• foreach(func)
  • 对数据集中每个元素使用函数func进行处理。