注：本文是跟着《Spark快速大数据分析》一书学习的笔记式总结，不喜勿喷。

RDD(Resilient Distributed Dataset)弹性分布式数据集是Spark对数据的核心抽象。RDD是一个不可变的分布式对象集合。每一个RDD都会被分成多个分区，不同的分区运行在集群的不同节点，这就构成了Spark的分布式计算模型。

RDD创建

我们可以通过两种方式创建RDD。一种方式是直接读取外部数据，这在我们实际使用中较常用，另一种是在驱动程序中分发驱动器程序中的对象集合（List或set），一般调试中会使用。

直接读取外部数据

JavaRDD<String> lines = sc.textFile("/path/to/README.md");
JavaRDD<String> lines1 = ctx.textFile("/path/to/README.md",1);//每次读取一行（类似Hadoop）

对一个集合进行并行化

JavaRDD<String> lines = sc.parallelize(Arrays.asList("pandas", "i like pandas"));

RDD操作

RDD支持两种操作：转化和行动。转化会生成一个新的RDD，如上面所说，RDD是不可变的，原来的RDD依然存在，只是在它的基础上，生成了一个新的RDD。行动会返回其他的数据类型，很多时候我们将其作为最后结果输出。实际上，转化操作是惰性的，只有在行动操作中，之前的转化操作才会真正执行。这可以避免多余的计算。

转化操作

许多转化操作都是针对各个元素的，也就是说，这些转化操作每次只会操作RDD 中的一个元素。不过并不是所有的转化操作都是这样的。
通过转化操作，从老的RDD中生成新的RDD,Spark会使用==谱系图==来记录不同的RDD之间的依赖关系。依靠谱系图，Spark可以在某些RDD计算出问题后，恢复出丢失的信息。

Java 版计算RDD 中各值的平方

JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1, 2, 3, 4));
JavaRDD<Integer> result = rdd.map(new Function<Integer, Integer>() {
public Integer call(Integer x) { return x*x; }
});
System.out.println(StringUtils.join(result.collect(), ","));

Java 中的flatMap() 将行数据切分为单词

JavaRDD<String> lines = sc.parallelize(Arrays.asList("hello world", "hi"));
JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
public Iterable<String> call(String line) {
return Arrays.asList(line.split(" "));
}
});
words.first(); // 返回"hello"

我们的RDD 中最常缺失的集合属性是元素的唯一性，因为常常有重复的元素。如果只
要唯一的元素，我们可以使用RDD.distinct() 转化操作来生成一个只包含不同元素的新
RDD。不过需要注意，distinct() 操作的开销很大，因为它需要将所有数据通过网络进行
混洗（shuffle），以确保每个元素都只有一份。

表：对一个数据为{1, 2, 3, 3}的RDD进行基本的RDD转化操作

表：对数据分别为{1, 2, 3}和{3, 4, 5}的RDD进行针对两个RDD的转化操作

行动操作

行动操作会把最终求得的结果返回到驱动器程序，或者写入外部存储系统中。它往往会触发转化RDD的运行。
调试小数据集的情况下，我们可以使用RDD的collection()函数，获取整个RDD中的数据。但这又一个前提，要单击内存能够成功存下这所有的数据才行。生产环境下这显然是不现实的。我们一般将数据写到HDFS或Amazon S3这种分布式的存储系统中。
Java 中的reduce()

Integer sum = rdd.reduce(new Function2<Integer, Integer, Integer>() {
public Integer call(Integer x, Integer y) { return x + y; }
});

Java 中的aggregate()

class AvgCount implements Serializable {
    public AvgCount(int total, int num) {
    this.total = total;
    this.num = num;
    }
   
    public int total;
    public int num;
    public double avg() {
        return total / (double) num;
    }
    }
    Function2<AvgCount, Integer, AvgCount> addAndCount =
    new Function2<AvgCount, Integer, AvgCount>() {
        public AvgCount call(AvgCount a, Integer x) {
            a.total += x;
            a.num += 1;
            return a;
        }
    };
    Function2<AvgCount, AvgCount, AvgCount> combine =
    new Function2<AvgCount, AvgCount, AvgCount>() {
        public AvgCount call(AvgCount a, AvgCount b) {
            a.total += b.total;
            a.num += b.num;
            return a;
        }
    };
    AvgCount initial = new AvgCount(0, 0);
    AvgCount result = rdd.aggregate(initial, addAndCount, combine);
    System.out.println(result.avg());

表：对一个数据为{1, 2, 3, 3}的RDD进行基本的RDD行动操作

持久化

对于Spark，虽然RDD是惰性求值的，但是，如果简单地对其调用行动操作，多次使用某一个RDD时就会重复计算。为了避免多次重新计算一个RDD,我们可以将其持久化。

表：org.apache.spark.storage.StorageLevel和pyspark.StorageLevel中的持久化级
别

我们在第一次对这个RDD 调用行动操作前就调用了persist() 方法。==persist() 调
用本身不会触发强制求值==。
RDD 还有一个方法叫作unpersist()，调用该方法可以手动把持久化的RDD 从缓
存中移除。

Java在不同RDD类型间转换

在Java 中，各种RDD 的特殊类型间的转换更为明确。Java 中有两个专门的类JavaDoubleRDD
和JavaPairRDD，来处理特殊类型的RDD，这两个类还针对这些类型提供了额外的函数。
这让你可以更加了解所发生的一切，但是也显得有些累赘。
要构建出这些特殊类型的RDD，需要使用特殊版本的类来替代一般使用的Function 类。如果
要从T 类型的RDD 创建出一个DoubleRDD，我们就应当在映射操作中使用DoubleFunction<T>
来替代Function<T, Double>。表3-5 展示了一些特殊版本的函数类及其用法。
此外，我们也需要调用RDD 上的一些别的函数（因此不能只是创建出一个DoubleFunction
然后把它传给map()）。当需要一个DoubleRDD 时，我们应当调用mapToDouble() 来替代
map()，跟其他所有函数所遵循的模式一样。
表3-5：Java中针对专门类型的函数接口

例：用Java 创建DoubleRDD

JavaDoubleRDD result = rdd.mapToDouble(
    new DoubleFunction<Integer>() {
        public double call(Integer x) {
        return (double) x * x;
        }
});
System.out.println(result.mean());

键值对RDD

众所周知，Hadoop的MapReduce一般处理键值对数据。Map输出，Reduce输入和输出都是Key-Value形式。在Spark中我们也支持键值对形式。并且，Spark的键值对Rdd相对MapReduce的键值对来说，操作更为简单，形式更加复杂多样。Spark 为包含键值对类型的RDD 提供了一些专有的操作。这些RDD 被称为pair RDD。

创建键值对RDD

Java 没有自带的二元组类型，因此Spark 的Java API 让用户使用scala.Tuple2 类来创建二
元组。这个类很简单：Java 用户可以通过new Tuple2(elem1, elem2) 来创建一个新的二元
组，并且可以通过._1() 和._2() 方法访问其中的元素。
在Java 中使用第一个单词作为键创建出一个pair RDD

PairFunction<String, String, String> keyData =
new PairFunction<String, String, String>() {
    public Tuple2<String, String> call(String x) {
       return new Tuple2(x.split(" ")[0], x);
    }
};
JavaPairRDD<String, String> pairs = lines.mapToPair(keyData);

表4-1：Pair RDD的转化操作（以键值对集合{(1, 2), (3, 4), (3, 6)}为例）

表4-2：针对两个pair RDD的转化操作（rdd = {(1, 2), (3, 4), (3, 6)}other = {(3, 9)}）

combineByKey()

combineByKey()是最为常用的基于键进行聚合的函数。大多数基于键聚合的函数都是用它
实现的。和aggregate() 一样，combineByKey() 可以让用户返回与输入数据的类型不同的
返回值。
要理解combineByKey()， 要先理解它在处理数据时是如何处理每个元素的。由于
combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就
和之前的某个元素的键相同。
如果这是一个新的元素，combineByKey() 会使用一个叫作createCombiner() 的函数来创建
那个键对应的累加器的初始值。需要注意的是，这一过程会在每个分区中第一次出现各个
键时发生，而不是在整个RDD 中第一次出现一个键时发生。
如果这是一个在处理当前分区之前已经遇到的键，它会使用mergeValue() 方法将该键的累
加器对应的当前值与这个新的值进行合并。
由于每个分区都是独立处理的，因此对于同一个键可以有多个累加器。如果有两个或者更
多的分区都有对应同一个键的累加器，就需要使用用户提供的mergeCombiners() 方法将各
个分区的结果进行合并。
如果已知数据在进行combineByKey() 时无法从map 端聚合中获益的话，可以
禁用它。例如，由于聚合函数（追加到一个队列）无法在map 端聚合时节约
任何空间，groupByKey() 就把它禁用了。如果希望禁用map 端组合，就需要
指定分区方式。就目前而言，你可以通过传递rdd.partitioner 来直接使用
源RDD 的分区方式。
例：在Java 中使用combineByKey() 求每个键对应的平均值

public static class AvgCount implements Serializable {
    public AvgCount(int total, int num) { total_ = total; num_ = num; }
        public int total_;
        public int num_;
        public float avg() { returntotal_/(float)num_; }
    }
    Function<Integer, AvgCount> createAcc = new Function<Integer, AvgCount>() {
        public AvgCount call(Integer x) {
            return new AvgCount(x, 1);
        }
    };
    Function2<AvgCount, Integer, AvgCount> addAndCount =
        new Function2<AvgCount, Integer, AvgCount>() {
        public AvgCount call(AvgCount a, Integer x) {
            a.total_ += x;
            a.num_ += 1;
            return a;
        }
    };
    Function2<AvgCount, AvgCount, AvgCount> combine =
    new Function2<AvgCount, AvgCount, AvgCount>() {
        public AvgCount call(AvgCount a, AvgCount b) {
            a.total_ += b.total_;
            a.num_ += b.num_;
            return a;
        }
    };
    AvgCount initial = new AvgCount(0,0);
    JavaPairRDD<String, AvgCount> avgCounts =
    nums.combineByKey(createAcc, addAndCount, combine);
    Map<String, AvgCount> countMap = avgCounts.collectAsMap();
    for (Entry<String, AvgCount> entry : countMap.entrySet()) {
    System.out.println(entry.getKey() + ":" + entry.getValue().avg());
}

Pair RDD的行动操作

表：Pair RDD的行动操作（以键值对集合{(1, 2), (3, 4), (3, 6)}为例）