spark RDD常用算子(一)
编程语言
2018-11-01 23:34:04
阅读次数: 0
- filter
- map
- flatMap
算法解释 有时候,我们希望对某个元素生成多个元素,实现该功能的操作叫作 flatMap() faltMap的函数应用于每一个元素,对于每一个元素返回的是多个元素组成的迭代器(想要了解更多,请参考scala的flatMap和map用法)
源文件 | 同上 |
scala代码 var flatMapLine = mspLine.flatMap(line => {
for (i <- 0 until line._2.length) yield (line._1, line._2, i)
}).first()
过滤结果 (2018-05-04,2018-05-04 09:39:27.286 [main] 14 ERROR - com.test.dao.Test - XmlFileLoader: 111,0)
- distinct
算法解释 distinct将RDD中的元素进行去重操作。图9中的每个方框代表一个RDD分区,通过distinct函数,将数据去重。 例如,重复数据V1、 V1去重后只保留一份V1。
源文件 var array = List(1, 2, 3, 4, 5, 6, 1, 1)
scala代码 var array = List(1, 2, 3, 4, 5, 6, 1, 1)
var rdd = sc.parallelize(array)
var rddDistinct = rdd.distinct()
println(rddDistinct.collect().mkString(","))
过滤结果 4,6,2,1,3,5
- union
算法解释 使用 union 函数时需要保证两个 RDD 元素的数据类型相同,返回的 RDD 数据类型和被合并的 RDD 元素数据类型相同,并不进行去重操作,保存所有元素。如果想去重可以使用 distinct()。同时 Spark 还提供更为简洁的使用 union 的 API,通过 ++ 符号相当于 union 函数操作
源文件 var array = List(1, 2, 3, 4, 5, 6, 1, 1)
var array2 = List(1, 2, 3, 4, 5, 6, 1, 1)
scala代码 var array = List(1, 2, 3, 4, 5, 6, 1, 1)
var rdd = sc.parallelize(array)
var array2 = List(1, 2, 3, 4, 5, 6, 1, 1)
var rdd2 = sc.parallelize(array)
var rddDistinct = rdd ++ rdd2 // or rdd.union(rdd2)
println(rddDistinct.collect().mkString(","))
过滤结果 1,2,3,4,5,6,1,1,1,2,3,4,5,6,1,1
- intersection
算法解释 该函数返回两个RDD的交集,并且去重。intersection 需要混洗数据,比较浪费性能
源文件 var array = List(1, 2, 3, 4, 5, 6, 1, 1)
var array2 = List(1, 2, 3, 4, 5, 6, 1, 1)
scala代码 var array = List(1, 2, 3, 4, 5, 6, 1, 1)
var rdd = sc.parallelize(array)
var array2 = List(1, 2, 3, 4, 5, 6, 1, 1)
var rdd2 = sc.parallelize(array)
var rddIntersection = rdd.intersection(rdd2)
println(rddIntersection .collect().mkString(","))
过滤结果 4,6,2,1,3,5
- subtract
算法解释 该函数类似于intersection,但返回在RDD中出现,并且不在otherRDD中出现的元素,不去重
源文件 var array = List(1, 2, 3, 4, 5, 6, 1, 1)
var array2 = List(1, 2, 3, 4, 5, 6, 1, 1)
scala代码 val rdd = sc.parallelize(Array(1, 2, 3, 4, 5))
val rdd2 = sc.parallelize(Array(1, 2, 3))
var substractRDD = rdd .subtract(rdd2)
println(substractRDD.collect().mkString(","))
过滤结果 4,5
- cartesian
算法解释 对 两 个 RDD 内 的 所 有 元 素 进 行 笛 卡 尔 积 操 作。 操 作 后, 内 部 实 现 返 回CartesianRDD。开销大 例 如: V1 和 另 一 个 RDD 中 的 W1、 W2、 Q5 进 行 笛 卡 尔 积 运 算 形 成 (V1,W1)、(V1,W2)、 (V1,Q5)
源文件 var array = List(1, 2, 3, 4, 5, 6, 1, 1)
var array2 = List(1, 2, 3, 4, 5, 6, 1, 1)
scala代码 val rdd = sc.parallelize(Array("a", "b", "c", "d", "e"))
val rdd2 = sc.parallelize(Array(1, 2, 3))
var cartesianRDD = rdd.cartesian(rdd2)
println(substractRDD.collect().mkString(","))
过滤结果 (a,1),(b,1),(a,2),(a,3),(b,2),(b,3),(c,1),(d,1),(e,1),(c,2),(c,3),(d,2),(d,3),(e,2),(e,3)
- mapToPair
算法解释 scala是没有mapToPair函数的,scala版本map。
源文件 val rdd = sc.parallelize(Array("a", "b", "c", "d", "e"))
scala代码 val rdd = sc.parallelize(Array("a", "b", "c", "d", "e"))
var mapToPairRDD = rdd.map(a => (a(0), 1))
println(mapToPairRDD.collect().mkString(","))
过滤结果 (a,1),(b,1),(c,1),(d,1),(e,1)
- flatMapToPair
算法解释 类似于xxx连接 mapToPair是一对一,一个元素返回一个元素,而flatMapToPair可以一个元素返回多个,相当于先flatMap,在mapToPair 例子: 将每一个单词都分成键为
源文件 val rdd = sc.parallelize(Array("aa", "bb", "cc", "dd", "ee"))
scala代码 val rdd = sc.parallelize(Array("aa", "bb", "cc", "dd", "ee"))
val flatRDD = rdd.flatMap(f => f).map(l => (l, 1))
println(flatRDD.collect().mkString(","))
过滤结果 (a,1),(a,1),(b,1),(b,1),(c,1),(c,1),(d,1),(d,1),(e,1),(e,1)
- flatMapToPair
算法解释 类似于xxx连接 mapToPair是一对一,一个元素返回一个元素,而flatMapToPair可以一个元素返回多个,相当于先flatMap,在mapToPair 例子: 将每一个单词都分成键为
源文件 val rdd = sc.parallelize(Array("aa", "bb", "cc", "dd", "ee"))
scala代码 val rdd = sc.parallelize(Array("aa", "bb", "cc", "dd", "ee"))
val flatRDD = rdd.flatMap(f => f).map(l => (l, 1))
println(flatRDD.collect().mkString(","))
过滤结果 (a,1),(a,1),(b,1),(b,1),(c,1),(c,1),(d,1),(d,1),(e,1),(e,1)
转载自 blog.csdn.net/u010020897/article/details/83541384