- 对RDD的学习进行一下整理
RDD:基于内存的集群计算容错抽象
分布式内存抽象的概念---弹性分布式数据集（RDD），它具备MapReduce等数据流模型的容错特性，并且允许开发人员在大型集群上执行基于内存的计算。
为了有效地实现容错，RDD提供了一种高度受限的共享内存，即RDD是只读的，并且只能通过其他 RDD上的批量操作来创建。
RDD基于工作集的应用，同时具有数据流模型的特点：自动容错、位置感知调度和可伸缩性。允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，提升了查询速度RDD提供了一种高度受限的共享内存模型，即 RDD是只读的记录分区的集合，只能通过在其他RDD执行确定的转换操作(如map、join 和group by)而创建，然而这些限制使得实现容错的开销很低。与分布式共享内存系统需要付出高昂代价的检查点和回滚机制不同，RDD通过Lineage来重建丢失的分区：一个RDD中包含了如何从其他 RDD衍生所必需的相关信息，从而不需要检查点操作就可以重构丢失的数据分区。
- 编程模型
a) 在Spark中，RDD被表示为对象，通过这些对象上的方法(或函数)调用转换。
b) 定义RDD后，可在动作中使用RDD。动作是向应用程序返回值，或向存储系统导出数据的那些操作。 _________使用RDD
- 如count(返回 RDD中的元素个数)，collect (返回元素本身)， save (将RDD输出到存储系统)
- 用户可以请求将 RDD缓存，这样运行时将已经计算好的 RDD分区存储起来，以加速后期的重用。缓存的RDD一般存储在内存中。RDD还允许用户根据关键字(key)指定分区顺序，这是一个可选的功能。目前支持哈希分区和范围分区
- 示例：控制台日志挖掘
1 lines = spark.textFile("hdfs://...")
2 errors = lines.filter(_.startsWith("ERROR"))
3 errors.cache()
- Spark编程接口
a) 要使用 Spark，开发者需要编写一个 driver 程序，连接到集群以运行 Worker（Worker是长时间运行的进程，将RDD分区以Java对象的形式缓存在内存中）
b) Spark中支持的RDD转换和动作
变换是对一个RDD进行操作得到一个新的RDD（延迟的）。
动作返回一个结果到驱动程序或者写入到存储并开始计算的操作。
返回类型：返回RDD的就是变换，反之，动作返回其他类型。
- Spark中的RDD，简单来说就是所有对象的一个不可变的分布式集合。每个都被分割为多个分区，这就可以在集群的不同节点上进行计算。RDD可以任何Python，Java，Scala对象类型，包括用户自定义类型。
- 创建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到磁盘上，而不是内存中也是可能的。
- 处理键值对
a) 创建
- 格式加载直接返回键值数据为pairRDD
- 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
- combineByKey()是针对每个键的更一般的聚合函数。大多数其他的对每个键合并的函数都是用它实现的, 和aggregate()一样，combinByKey()允许用户返回和输入数据类型不同的值。
- combineByKey()会遍历一个分区中的所有元素。如果是新元素，combineByKey()会创建combiner()，也就是我们提供给它的函数来为这个新的键创建初值用于累积计算。（这个第一次是每个分区中新的键的第一次出）
- 示例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(_))
- 数据分组
a) 有主键的数据的惯用法是根据键来对数据分组。groupByKey()就可以用RDD中的主键对我们的数据进行分组。对于一个由类型为K的主键和类型为V的值组成的RDD，groupByKey()后我们得到的RDD类型为[K, Iterable[V]]。
- 连接
a) 得到了有主键的数据后，一个最有用的操作就是可以和其他有主键的数据一起使用。对pair RDD来说，连接数据到一起恐怕是最常见的操作。Spark支持全系列的连接操作，包括右外连接，左外连接，交叉连接和内连接。
- 内连接
- 示例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上可用的动作