04、常用RDD操作

函数名

目的

示例

结果

RDD[U] map(f: T => U)

T：初始元素类型

U：转换后元素类型

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

rdd.map(x => x + 1)

测试数据：{1, 2, 3, 3}

输出结果：{2, 3, 4, 4}

RDD[U] flatMap(f: T => TraversableOnce[U])

TraversableOnce：特质，具有集合与迭代器特性

将函数应用于 RDD 中的每个元素，将返回的迭代器的所有内容构成新的 RDD。通常用来切分单词

rdd.flatMap(x => x.to(3))

测试数据：{1, 2, 3, 3}

输出结果：{1, 2, 3, 2, 3, 3, 3}

RDD[T] filter(f: T => Boolean)

返回一个由通过传给 filter()的函数的元素组成的 RDD

rdd.filter(x => x != 1)

测试数据：{1, 2, 3, 3}

输出结果：{2, 3, 3}

RDD[T] distinct()

去重

rdd.distinct()

测试数据：{1, 2, 3, 3}

输出结果：{1, 2, 3}

RDD[(K, Iterable[V])] gropuByKey()

K：key类型

V：value类型

根据key进行分组，返回的元素每个key对应一个Iterable<value>

测试数据：

("class1", 80),

("class2", 75),     

("class1", 90),

("class2", 60)

输出结果：

("class1",[80,90])

("class2",[75,60])

RDD[(K, V)] reduceByKey(func: (V, V) => V)

对每个key对应的value进行reduce操作

rdd. reduceByKey(_ + _)

测试数据：

("class1", 80),

("class2", 75),     

("class1", 90),

("class2", 60)

输出结果：

("class1", 170),

("class2", 135)

RDD[(K, V)] sortByKey()

根据key进行排序操作

测试数据：

(65, "leo")

(50, "tom")

(100, "marry")

(85, "jack")

输出结果：

(100, "marry")

(85, "jack")

(65, "eo")

(50, "tom")

RDD[T] sortBy(f: (T) => K,

      ascending: Boolean,

      numPartitions: Int)

既可以根据key也可以根据value进行排序，传入的是一个(T) => K 转换函数

rdd.sortBy(_._2, false, 1)

ascending：false时降序

测试数据：

("leo", 65),

("tom", 50),

("marry", 100),

("jack", 80)

输出结果：

("marry", 100),

("jack", 80)

("leo", 65),

("leo", 65),

join

对两个包含<key,value>对的RDD进行join操作，返回类型<key,Tuple2(key,value)>

cogroup

同join，但是每个key对应的Iterable<value>都会传入自定义函数进行处理

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

union()

两个RDD联合

rdd.union(other)

{1, 2, 3, 3, 4, 5}

intersection()

两个RDD 交集

rdd.intersection(other)

{3}

subtract()

两个RDD相减

rdd.subtract(other)

{1, 2}

cartesian()

两个RDD相减笛卡儿积

rdd.cartesian(other)

{(1, 3), (1, 4), ...(3, 5)}

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

函数名

目的

示例

结果

collect()

返回 RDD 中的所有元素

rdd.collect()

{1, 2, 3, 3}

count()

RDD 中的元素个数

rdd.count()

4

countByValue()

各元素在 RDD 中出现的次数

rdd.countByValue()

{(1, 1),(2, 1),(3, 2)}

take(num)

从 RDD 中返回 num 个元素

rdd.take(2)

{1, 2}

top(num)

从 RDD 中返回最前面的 num个元素

rdd.top(2)

{3, 3}

takeOrdered(num)(ordering)

从 RDD 中按照提供的顺序返回最前面的 num 个元素

rdd.takeOrdered(2)(myOrdering)

{3, 3}

takeSample(withReplace

ment, num, [seed])

从 RDD 中返回任意一些元素

rdd.takeSample(false, 1)

非确定的

reduce(func)

并 行 整 合 RDD 中 所 有 数 据

（例如 sum）

rdd.reduce((x, y) => x + y)

9

fold(zero)(func)

和 reduce() 一 样， 但 是 需 要

提供初始值

rdd.fold(0)((x, y) => x + y)

9

aggregate(zeroValue)

(seqOp, combOp)

和 reduce() 相 似， 但 是 通 常

返回不同类型的函数

rdd.aggregate((0, 0))

((x, y) =>

(x._1 + y, x._2 + 1),

(x, y) =>

(x._1 + y._1, x._2 + y._2))

(9,4)

foreach(func)

对 RDD 中的每个元素使用给

定的函数

rdd.foreach(func)

无