Spark RDD

时间:2022-01-12 16:44:21
  1. 对RDD的学习进行一下整理

RDD:基于内存的集群计算容错抽象

分布式内存抽象的概念---弹性分布式数据集(RDD),它具备MapReduce等数据流模型的容错特性,并且允许开发人员在大型集群上执行基于内存的计算。

为了有效地实现容错,RDD提供了一种高度受限的共享内存,即RDD是只读的,并且只能通过其他 RDD上的批量操作来创建。

RDD基于工作集的应用,同时具有数据流模型的特点:自动容错、位置感知调度和可伸缩性。允许用户在执行多个查询时显式地将工作集缓存在内存中,后续的查询能够重用工作集,提升了查询速度RDD提供了一种高度受限的共享内存模型,即 RDD是只读的记录分区的集合,只能通过在其他RDD执行确定的转换操作(如map、join 和group by)而创建,然而这些限制使得实现容错的开销很低。与分布式共享内存系统需要付出高昂代价的检查点和回滚机制不同,RDD通过Lineage来重建丢失的分区:一个RDD中包含了如何从其他 RDD衍生所必需的相关信息,从而不需要检查点操作就可以重构丢失的数据分区。

  1. 编程模型

a) 在Spark中,RDD被表示为对象,通过这些对象上的方法(或函数)调用转换。

b) 定义RDD后,可在动作中使用RDD。动作是向应用程序返回值,或向存储系统导出数据的那些操作。   _________使用RDD

  1. 如count(返回 RDD中的元素个数),collect (返回元素本身),       save (将RDD输出到存储系统)
  2. 用户可以请求将 RDD缓存,这样运行时将已经计算好的 RDD分区存储起来,以加速后期的重用。缓存的RDD一般存储在内存中。RDD还允许用户根据关键字(key)指定分区顺序,这是一个可选的功能。目前支持哈希分区和范围分区
  3. 示例:控制台日志挖掘

1    lines   =   spark.textFile("hdfs://...")

2 errors  =   lines.filter(_.startsWith("ERROR"))

3 errors.cache()

  1. Spark编程接口

a) 要使用 Spark,开发者需要编写一个 driver 程序,连接到集群以运行 Worker(Worker是长时间运行的进程,将RDD分区以Java对象的形式缓存在内存中)

b) Spark中支持的RDD转换和动作

变换是对一个RDD进行操作得到一个新的RDD(延迟的)。

动作返回一个结果到驱动程序或者写入到存储并开始计算的操作。

返回类型:返回RDD的就是变换,反之,动作返回其他类型。

  1. Spark中的RDD,简单来说就是所有对象的一个不可变的分布式集合。每个都被分割为多个分区,这就可以在集群的不同节点上进行计算。RDD可以任何Python,Java,Scala对象类型,包括用户自定义类型。
  2. 创建RDD:

a) 加载一个外部数据集

val  rdd1 = sc.textFile ("d:/e.log")

b) 在驱动程序中分发一个对象集合(如list或set)

c) 在任何时候你都可以定义一个新的RDD,但是Spark总是以一种lazy的方式计算它们, 最后,每次你执行个动作,Spark的RDD默认会被重新计算.

d) 重用RDD,RDD.persist()要求Spark对RDD持久化,在初次计算之后,Spark 可以保存RDD的内容到内存中(在你的集群中跨机器分区),并在未来的动作中重用。持久化RDD到磁盘上,而不是内存中也是可能的。

  1. 处理键值对

a) 创建

  1. 格式加载直接返回键值数据为pairRDD
  2. Map()操作返回键值对。

val pairs = lines.map(x => (x.split(" ")(0), x))

b) 对Pair RDD的变换

示例4-5 Scala中对第二个元素简单过滤

pairs.filter{case (key, value) => value.length < 20}

有时我们只是想访问pair RDD的值那部分,处理pair就会有些不灵活。由于这是很常见的模式,所以Spark提供了mapValues(func)函数,等同于

map{case (x, y)=> (x, func(y))}。

c) 聚合

Spark提供了按相同的键合并的一组类似的操作reduceByKey(),运行多个reduce()操作,数据集中的每个键对应一个,以便对相同的键进行合并,因为数据集可以有非常多的键,所以reduceByKey()没有实现成返回一个值到用户程序的动作,而是返回一个由每个键及对应合并后的值组成的新RDD。foldByKey()和fold()十分相似,都有一个和RDD中数据同类型的零值和合并函数。

 

示例4-7. Scala中用reduceByKey()和mapValues计算每个键的均值

rdd.mapValues(x => (x, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))

示例4-10. Scala中的Word count

val input = sc.textFile("s3://...")

val words = input.flatMap(x => x.split(" "))

val result = words.map(x => (x, 1)).reduceByKey((x, y) => x + y)

我们可以对第一个RDD使用countByValue()更快的实现单词计数:

input.flatMap(x => x.split(“ ”)).countByValue()

d) combineByKey

  1. combineByKey()是针对每个键的更一般的聚合函数。大多数其他的对每个键合并的函数都是用它实现的, 和aggregate()一样,combinByKey()允许用户返回和输入数据类型不同的值。
  2. combineByKey()会遍历一个分区中的所有元素。如果是新元素,combineByKey()会创建combiner(),也就是我们提供给它的函数来为这个新的键创建初值用于累积计算。(这个第一次是每个分区中新的键的第一次出)
  3. 示例4-13. Scala中使用combineByKey()计算每个键的平均值

val result = input.combineByKey(

(v) => (v, 1),

(acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),

(acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 +acc2._2)).map{ case(key,value)=>(key,value._1/value._2.toFloat) }

result.collectAsMap().map(println(_))

  1. 数据分组

a) 有主键的数据的惯用法是根据键来对数据分组。groupByKey()就可以用RDD中的主键对我们的数据进行分组。对于一个由类型为K的主键和类型为V的值组成的RDD,groupByKey()后我们得到的RDD类型为[K, Iterable[V]]。

  1. 连接

a) 得到了有主键的数据后,一个最有用的操作就是可以和其他有主键的数据一起使用。对pair RDD来说,连接数据到一起恐怕是最常见的操作。Spark支持全系列的连接操作,包括右外连接,左外连接,交叉连接和内连接。

  1. 内连接
    1. 示例4-17. Scala shell中进行内连接

storeAddress = {

(Store("Ritual"),"1026 Valencia St"),

(Store("Philz"),"748 Van Ness Ave"),

(Store("Philz"),"3101 24th St"),

(Store("Starbucks"),"Seattle")}

storeRating = {

(Store("Ritual"), 4.9),

(Store("Philz"), 4.8))}

storeAddress.join(storeRating) == {

(Store("Ritual"), ("1026 Valencia St", 4.9)),

(Store("Philz"), ("748 Van Ness Ave", 4.8)),

(Store("Philz"), ("3101 24th St", 4.8))

}

leftOuterJoin(),结果RDD中的每个主键都在源RDD中存在

示例4-18. leftOuterJoin() 和rightOuterJoin()

storeAddress.leftOuterJoin(storeRating) ==

{(Store("Ritual"),("1026 Valencia St",Some(4.9))),

(Store("Starbucks"),("Seattle",None)),

(Store("Philz"),("748 Van Ness Ave",Some(4.8))),

(Store("Philz"),("3101 24th St",Some(4.8)))}

storeAddress.rightOuterJoin(storeRating) ==

{(Store("Ritual"),(Some("1026 Valencia St"),4.9)),

(Store("Philz"),(Some("748 Van Ness Ave"),4.8)),

(Store("Philz"), (Some("3101 24th St"),4.8))}

Pair RDD上可用的动作