Spark2.3.0 RDD操作

smartsi

发布于 2019-08-07 15:05:40

2.3K0

发布于 2019-08-07 15:05:40

文章被收录于专栏：SmartSiSmartSi

RDD支持两种类型的操作：

转换操作(transformations): 从现有数据集创建一个新数据集
动作操作(actions): 在数据集上进行计算后将值返回给驱动程序

例如，map 是一个转换操作，传递给每个数据集元素一个函数并返回一个新 RDD 表示返回结果。另一方面，reduce 是一个动作操作，使用一些函数聚合 RDD 的所有元素并将最终结果返回给驱动程序（尽管还有一个并行的 reduceByKey 返回一个分布式数据集）。

在 Spark 中，所有的转换操作(transformations)都是惰性(lazy)的，它们不会马上计算它们的结果。相反，它们仅仅记录应用到基础数据集(例如一个文件)上的转换操作。只有当 action 操作需要返回一个结果给驱动程序的时候，转换操作才开始计算。

这个设计能够让 Spark 运行得更加高效。例如，我们知道：通过 map 创建的新数据集将在 reduce 中使用，并且仅仅返回 reduce 的结果给驱动程序，而不必将比较大的映射后的数据集返回。

1. 基础

为了说明 RDD 基础知识，请考虑以下简单程序：

Java版本:

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
int totalLength = lineLengths.reduce((a, b) -> a + b);

Scala版本:

val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

Python版本:

lines = sc.textFile("data.txt")
lineLengths = lines.map(lambda s: len(s))
totalLength = lineLengths.reduce(lambda a, b: a + b)

第一行定义了一个来自外部文件的基本 RDD。这个数据集并未加载到内存中或做其他处理：lines 仅仅是一个指向文件的指针。第二行将 lineLengths 定义为 map 转换操作的结果。其次，由于转换操作的惰性(lazy)，lineLengths 并没有立即计算。最后，我们运行 reduce，这是一个动作操作。此时，Spark 把计算分成多个任务(task)，并让它们运行在多台机器上。每台机器都运行 map 的一部分以及本地 reduce。然后仅仅将结果返回给驱动程序。

如果稍后还会再次使用 lineLength，我们可以在运行 reduce 之前添加：

Java版本:

lineLengths.persist(StorageLevel.MEMORY_ONLY());

Scala版本:

lineLengths.persist()

Python版本:

lineLengths.persist()

这将导致 lineLength 在第一次计算之后被保存在内存中。

2. 传递函数给Spark

Spark 的 API 很大程度上依赖于运行在集群上的驱动程序中的函数。

2.1 Java版本

在 Java 中，函数由 org.apache.spark.api.java.function 接口实现。创建这样的函数有两种方法：

在你自己类中实现 Function 接口，作为匿名内部类或命名内部类，并将其实例传递给Spark。
使用 lambda 表达式来简洁地定义一个实现。

虽然本指南的大部分内容都使用 lambda 语法进行简明说明，但很容易以长格式使用所有相同的API。例如，我们可以按照以下方式编写我们的代码：

匿名内部类

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new Function<String, Integer>() {
  public Integer call(String s) { return s.length(); }
});
int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
  public Integer call(Integer a, Integer b) { return a + b; }
});

或者命名内部类

class GetLength implements Function<String, Integer> {
  public Integer call(String s) { return s.length(); }
}
class Sum implements Function2<Integer, Integer, Integer> {
  public Integer call(Integer a, Integer b) { return a + b; }
}

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new GetLength());
int totalLength = lineLengths.reduce(new Sum());

请注意，Java中的匿名内部类也可以访问封闭范围内的变量，只要它们标记为final。 Spark会将这些变量的副本发送给每个工作节点，就像其他语言一样。

2.2 Scala版本

有两种推荐的的方法可以做到这一点：

匿名函数语法，可用于短片段代码。
全局单例对象中的静态方法。例如，您可以定义对象 MyFunctions，然后传递 MyFunctions.func1，如下所示：

object MyFunctions {
  def func1(s: String): String = { ... }
}

myRdd.map(MyFunctions.func1)

虽然也可以在类实例中传递方法的引用（与单例对象相反），但这需要将包含该类的对象与方法一起发送。例如，考虑：

class MyClass {
  def func1(s: String): String = { ... }
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}

在这里，如果我们创建一个新的 MyClass 实例并调用 doStuff 方法，那么其中的 map 会引用该 MyClass 实例的 func1 方法，因此需要将整个对象发送到集群。它类似于编写 rdd.map（x => this.func1（x）） 。

以类似的方式，访问外部对象的字段将引用整个对象：

class MyClass {
  val field = "Hello"
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}

等价于 rdd.map（x => this.field + x），它引用了 this 对象的所有东西。为了避免这个问题，最简单的方法是将字段复制到本地变量中，而不是从外部访问它：

def doStuff(rdd: RDD[String]): RDD[String] = {
  val field_ = this.field
  rdd.map(x => field_ + x)
}

3. 使用键值对

虽然大多数 Spark 操作可以在任意类型对象的 RDD 上工作，但是还是几个特殊操作只能在键值对的 RDD 上使用。最常见的是分布式 shuffle 操作，例如按键分组或聚合元素。

3.1 Java版本

在 Java 中，使用 Scala 标准库中的 scala.Tuple2 类来表示键值对。可以如下简单地调用 new Tuple2（a，b） 来创建一个元组，然后用 tuple._1（） 和 tuple._2（） 访问它的字段。

键值对的 RDD 由 JavaPairRDD 类表示。你可以使用特殊版本的 map 操作（如 mapToPair 和 flatMapToPair）从 JavaRDD 来构建 JavaPairRDD。JavaPairRDD 具有标准的 RDD 函数以及特殊的键值对函数。

例如，以下代码在键值对上使用 reduceByKey 操作来计算每行文本在文件中的出现次数：

JavaRDD<String> lines = sc.textFile("data.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(s -> new Tuple2(s, 1));
JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);

例如，我们也可以使用 counts.sortByKey（） 来按字母顺序来对键值对排序，最后使用 counts.collect（） 将结果作为对象数组返回到驱动程序。

3.2 Scala版本

在 Scala 中，这些操作在包含 Tuple2 对象的 RDD 上可以自动获取（内置元组，通过简单写入（a，b）创建）。在 PairRDDFunctions 类中提供了键值对操作，该类自动包装元组的 RDD。

例如，以下代码在键值对上使用 reduceByKey 操作来计算每行文本在文件中的出现次数：

val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)

例如，我们也可以使用 counts.sortByKey（） 来按字母顺序来对键值对排序，最后使用 counts.collect（） 将结果作为对象数组返回到驱动程序。

在键值对操作时使用一个自定义对象作为 key 的时候，你需要确保自定义 equals() 方法和 hashCode() 方法是匹配的。更加详细的内容，查看 Object.hashCode()) 文档)中的契约概述。

4. 转换操作

下面列出了Spark支持的一些常见转换函数。有关详细信息，请参阅RDD API文档（Scala，Java，Python，R）和RDD函数doc（Scala，Java）。

4.1 map(func) 映射

将函数应用于 RDD 中的每个元素，将返回值构成新的 RDD。

List<String> aList = Lists.newArrayList("a", "B", "c", "b");
JavaRDD<String> rdd = sc.parallelize(aList);
// 小写转大写
JavaRDD<String> upperLinesRDD = rdd.map(new Function<String, String>() {
    @Override
    public String call(String str) throws Exception {
        if (StringUtils.isBlank(str)) {
            return str;
        }
        return str.toUpperCase();
    }
});
// A B C B

4.2 filter(func) 过滤

返回通过选择那些 func 函数返回 true 的元素形成一个新的 RDD。

List<String> list = Lists.newArrayList("a", "B", "c", "b");
JavaRDD<String> rdd = sc.parallelize(list);
// 只返回以a开头的字符串
JavaRDD<String> filterRDD = rdd.filter(new Function<String, Boolean>() {
    @Override
    public Boolean call(String str) throws Exception {
        return !str.startsWith("a");
    }
});
// B c b

4.3 flatMap(func) 一行转多行

类似于 map 函数，但是每个输入项可以映射为0个输出项或更多输出项（所以 func 应该返回一个序列而不是一个条目）。

List<String> list = Lists.newArrayList("a 1", "B 2");
JavaRDD<String> rdd = sc.parallelize(list);
// 一行转多行 以空格分割
JavaRDD<String> resultRDD = rdd.flatMap(new FlatMapFunction<String, String>() {
    @Override
    public Iterator<String> call(String s) throws Exception {
        if (StringUtils.isBlank(s)) {
            return null;
        }
        String[] array = s.split(" ");
        return Arrays.asList(array).iterator();
    }
});
// a
// 1
// B
// 2

4.4 distinct([numTasks]))

去重

List<String> aList = Lists.newArrayList("1", "3", "2", "3");
JavaRDD<String> aRDD = sc.parallelize(aList);
// 去重
JavaRDD<String> rdd = aRDD.distinct(); // 1 2 3

4.5 union(otherDataset) 并集

生成一个包含两个 RDD 中所有元素的 RDD。如果输入的 RDD 中有重复数据，union() 操作也会包含这些重复的数据．

List<String> aList = Lists.newArrayList("1", "2", "3");
List<String> bList = Lists.newArrayList("3", "4", "5");
JavaRDD<String> aRDD = sc.parallelize(aList);
JavaRDD<String> bRDD = sc.parallelize(bList);
// 并集
JavaRDD<String> rdd = aRDD.union(bRDD); // 1 2 3 3 4 5

4.6 intersection(otherDataset) 交集

求两个 RDD 共同的元素的 RDD。 intersection() 在运行时也会去掉所有重复的元素，尽管 intersection() 与 union() 的概念相似，但性能却差的很多，因为它需要通过网络混洗数据来发现共同的元素。

List<String> aList = Lists.newArrayList("1", "2", "3");
List<String> bList = Lists.newArrayList("3", "4", "5");
JavaRDD<String> aRDD = sc.parallelize(aList);
JavaRDD<String> bRDD = sc.parallelize(bList);
// 交集
JavaRDD<String> rdd = aRDD.intersection(bRDD); // 3

4.7 subtract(otherDataset) 差集

subtract 接受另一个 RDD 作为参数，返回一个由只存在第一个 RDD 中而不存在第二个 RDD 中的所有元素组成的 RDD

List<String> aList = Lists.newArrayList("1", "2", "3");
List<String> bList = Lists.newArrayList("3", "4", "5");
JavaRDD<String> aRDD = sc.parallelize(aList);
JavaRDD<String> bRDD = sc.parallelize(bList);
// 差集
JavaRDD<String> rdd = aRDD.subtract(bRDD); // 1 2

4.8 groupByKey 分组

根据键值对 key 进行分组。在（K，V）键值对的数据集上调用时，返回（K，Iterable ）键值对的数据集。

如果分组是为了在每个 key 上执行聚合（如求总和或平均值），则使用 reduceByKey 或 aggregateByKey 会有更好的性能。默认情况下，输出中的并行级别取决于父 RDD 的分区数。你可以传递一个可选参数 numTasks 来设置任务数量。

Tuple2<String, Integer> t1 = new Tuple2<String, Integer>("Banana", 10);
Tuple2<String, Integer> t2 = new Tuple2<String, Integer>("Pear", 5);
Tuple2<String, Integer> t3 = new Tuple2<String, Integer>("Banana", 9);
Tuple2<String, Integer> t4 = new Tuple2<String, Integer>("Apple", 4);
List<Tuple2<String, Integer>> list = Lists.newArrayList();
list.add(t1);
list.add(t2);
list.add(t3);
list.add(t4);
JavaPairRDD<String, Integer> rdd = sc.parallelizePairs(list);
// 分组
JavaPairRDD<String, Iterable<Integer>> groupRDD = rdd.groupByKey();

// Apple --- 4
// Pear --- 5
// Banana --- 10 9

4.9 reduceByKey(func, [numTasks]) 根据key聚合

当在（K，V）键值对的数据集上调用时，返回（K，V）键值对的数据集，使用给定的reduce函数 func 聚合每个键的值，该函数类型必须是（V，V）=> V。类似于 groupByKey，可以通过设置可选的第二个参数来配置reduce任务的数量。

Tuple2<String, Integer> t1 = new Tuple2<String, Integer>("Banana", 10);
Tuple2<String, Integer> t2 = new Tuple2<String, Integer>("Pear", 5);
Tuple2<String, Integer> t3 = new Tuple2<String, Integer>("Banana", 9);
Tuple2<String, Integer> t4 = new Tuple2<String, Integer>("Apple", 4);
List<Tuple2<String, Integer>> list = Lists.newArrayList();
list.add(t1);
list.add(t2);
list.add(t3);
list.add(t4);
JavaPairRDD<String, Integer> rdd = sc.parallelizePairs(list);
// 分组计算
JavaPairRDD<String, Integer> reduceRDD = rdd.reduceByKey(new Function2<Integer, Integer, Integer>() {
    @Override
    public Integer call(Integer v1, Integer v2) throws Exception {
        return v1 + v2;
    }
});

// Apple --- 4
// Pear --- 5
// Banana --- 19

4.10 sortByKey([ascending], [numPartitions])

根据key进行排序。在（K，V）键值对的数据集调用，其中 K 实现 Ordered 接口，按照升序或降序顺序返回按键排序的（K，V）键值对的数据集。

Tuple2<String, Integer> t1 = new Tuple2<String, Integer>("Banana", 10);
Tuple2<String, Integer> t2 = new Tuple2<String, Integer>("Pear", 5);
Tuple2<String, Integer> t3 = new Tuple2<String, Integer>("Apple", 4);
List<Tuple2<String, Integer>> list = Lists.newArrayList();
list.add(t1);
list.add(t2);
list.add(t3);
JavaPairRDD<String, Integer> rdd = sc.parallelizePairs(list);
// 根据key排序
JavaPairRDD<String, Integer> sortRDD = rdd.sortByKey(true);
// Apple --- 4
// Banana --- 10
// Pear --- 5

4.11 coalesce(numPartitions) 合并分区

对 RDD 分区进行合并，合并后分区数目为 numPartitions。大型数据集过滤之后可以对高效地运行操作很有帮助。

List<String> list = Lists.newArrayList("1", "2", "3", "4");
JavaRDD<String> rdd = sc.parallelize(list);
JavaRDD<String> coalesceRDD = rdd.coalesce(2, false);
System.out.println("分区个数:" + coalesceRDD.partitions().size());
// 分区个数:1

JavaRDD<String> coalesceRDD2 = rdd.coalesce(2, true);
System.out.println("分区个数:" + coalesceRDD2.partitions().size());
// 分区个数:2

如果可选参数 shuff 为 false 时，传入的参数大于现有的分区数目，RDD 的分区数不变，也就是说不经过shuffle，是无法将 RDD 的分区数增大。

4.12 repartition(numPartitions) 重新分区

对 RDD 中的数据重新洗牌来重新分区，分区数目可以增大也可以减少，并在各分区之间进行数据平衡。这总是通过网络混洗所有数据。

List<String> list = Lists.newArrayList("1", "2", "3", "4");
JavaRDD<String> rdd = sc.parallelize(list);
JavaRDD<String> coalesceRDD = rdd.repartition(2);
System.out.println("分区个数:" + coalesceRDD.partitions().size());
// 分区个数:2

我们从源码看到 repartition 只是 coalesce(numPartitions, shuffle = true) 的一个简易实现：

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
  coalesce(numPartitions, shuffle = true)
}

4.13 cartesian(otherDataset) 笛卡尔积

对两个 RDD 中的所有元素进行笛卡尔积操作。在类型为 T 和 U 的两个数据集上调用时，返回（T，U）键值对（所有元素对）数据集。

List<String> aList = Lists.newArrayList("1", "2");
List<String> bList = Lists.newArrayList("3", "4");
JavaRDD<String> aRDD = sc.parallelize(aList);
JavaRDD<String> bRDD = sc.parallelize(bList);
// 笛卡尔积
JavaPairRDD<String, String> cartesianRDD = aRDD.cartesian(bRDD);
// (1, 3)
// (1, 4)
// (2, 3)
// (2, 4)

4.14 cogroup(otherDataset, [numPartitions])

在类型（K，V）和（K，W）两个数据集上调用时，返回（K，（Iterable ，Iterable ））元组的数据集。这个操作也被称为 groupWith。

Tuple2<String, Integer> t1 = new Tuple2<String, Integer>("Banana", 10);
Tuple2<String, Integer> t2 = new Tuple2<String, Integer>("Pear", 5);
Tuple2<String, Integer> t3 = new Tuple2<String, Integer>("Apple", 4);

Tuple2<String, Integer> t4 = new Tuple2<String, Integer>("Banana", 2);
Tuple2<String, Integer> t5 = new Tuple2<String, Integer>("Apple", 11);
Tuple2<String, Integer> t6 = new Tuple2<String, Integer>("Banana", 7);
List<Tuple2<String, Integer>> list1 = Lists.newArrayList();
list1.add(t1);
list1.add(t2);
list1.add(t3);
List<Tuple2<String, Integer>> list2 = Lists.newArrayList();
list2.add(t4);
list2.add(t5);
list2.add(t6);

JavaPairRDD<String, Integer> rdd1 = sc.parallelizePairs(list1);
JavaPairRDD<String, Integer> rdd2 = sc.parallelizePairs(list2);

JavaPairRDD<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> cogroupRDD = rdd1.cogroup(rdd2);
cogroupRDD.foreach(new VoidFunction<Tuple2<String, Tuple2<Iterable<Integer>, Iterable<Integer>>>>() {
    @Override
    public void call(Tuple2<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> group) throws Exception {
        String key = group._1;
        Tuple2<Iterable<Integer>, Iterable<Integer>> value = group._2;
        System.out.println(key + " --- " + value.toString());
    }
});
// Apple --- ([4],[11])
// Pear --- ([5],[])
// Banana --- ([10],[2, 7])

5. 动作操作 (Action)

下面列出了Spark支持的一些常见操作。

5.1 reduce

接收一个函数作为参数，这个函数要操作两个相同元素类型的RDD并返回一个同样类型的新元素．

List<String> aList = Lists.newArrayList("aa", "bb", "cc", "dd");
JavaRDD<String> rdd = sc.parallelize(aList);
String result = rdd.reduce(new Function2<String, String, String>() {
    @Override
    public String call(String v1, String v2) throws Exception {
        return v1 + "#" + v2;
    }
});
System.out.println(result); // aa#bb#cc#dd

5.2 collect

将整个RDD的内容返回．

List<String> list = Lists.newArrayList("aa", "bb", "cc", "dd");
JavaRDD<String> rdd = sc.parallelize(list);
List<String> collect = rdd.collect();
System.out.println(collect); // [aa, bb, cc, dd]

5.3 take(n)

返回 RDD 中的n个元素，并且尝试只访问尽量少的分区，因此该操作会得到一个不均衡的集合．需要注意的是，这些操作返回元素的顺序与你的预期可能不一样．

List<String> list = Lists.newArrayList("aa", "bb", "cc", "dd");
JavaRDD<String> rdd = sc.parallelize(list);
List<String> collect = rdd.take(3);
System.out.println(collect); // [aa, bb, cc]

5.4 takeSample

有时需要在驱动器程序中对我们的数据进行采样，takeSample(withReplacement, num, seed) 函数可以让我们从数据中获取一个采样，并指定是否替换．

5.5 saveAsTextFile(path)

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

List<String> list = Lists.newArrayList("aa", "bb", "cc", "dd");
JavaRDD<String> rdd = sc.parallelize(list);
rdd.saveAsTextFile("/home/xiaosi/output");

输出:

xiaosi@ying:~/output$ cat *
aa
bb
cc
dd

5.6 saveAsSequenceFile(path)

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

5.7 foreach(func)

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

修改foreach（）之外的变量而不是累加器可能会导致未定义的行为。有关更多详细信息，请参阅了解闭包 Spark版本:2.3.0

原文：http://spark.apache.org/docs/2.3.0/rdd-programming-guide.html#rdd-operations

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2018-03-12，如有侵权请联系 cloudcommunity@tencent.com 删除

spark

java

scala

mapreduce

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度