2018-11-17 Spark算子练习常用Transformation(即转换,延迟加载)通过并行化scala集合创建RDD查看该rdd的分区数量union求并集,注意类型要一致intersecti

常用Transformation(即转换,延迟加载)

通过并行化scala集合创建RDD

val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))

查看该rdd的分区数量

rdd1.partitions.length

val rdd1 = sc.parallelize(List(5,6,4,7,3,8,2,9,1,10)) val rdd2 = sc.parallelize(List(5,6,4,7,3,8,2,9,1,10)).map(*2).sortBy(x=>x,true) val rdd3 = rdd2.filter(>10) val rdd2 = sc.parallelize(List(5,6,4,7,3,8,2,9,1,10)).map(*2).sortBy(x=>x+"",true) val rdd2 = sc.parallelize(List(5,6,4,7,3,8,2,9,1,10)).map(*2).sortBy(x=>x.toString,true)

val rdd4 = sc.parallelize(Array("a b c", "d e f", "h i j")) rdd4.flatMap(_.split(' ')).collect

val rdd5 = sc.parallelize(List(List("a b c", "a b b"),List("e f g", "a f g"), List("h i j", "a a b")))

List("a b c", "a b b") =List("a","b",))

rdd5.flatMap(.flatMap(.split(" "))).collect

union求并集,注意类型要一致

val rdd6 = sc.parallelize(List(5,6,4,7)) val rdd7 = sc.parallelize(List(1,2,3,4)) val rdd8 = rdd6.union(rdd7) rdd8.distinct.sortBy(x=>x).collect

intersection求交集

val rdd9 = rdd6.intersection(rdd7)

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 2), ("kitty", 3))) val rdd2 = sc.parallelize(List(("jerry", 9), ("tom", 8), ("shuke", 7), ("tom", 2)))

join(连接)

val rdd3 = rdd1.join(rdd2) val rdd3 = rdd1.leftOuterJoin(rdd2) val rdd3 = rdd1.rightOuterJoin(rdd2)

groupByKey

val rdd3 = rdd1 union rdd2 rdd3.groupByKey //(tom,CompactBuffer(1, 8, 2)) rdd3.groupByKey.map(x=>(x._1,x.2.sum)) groupByKey.mapValues(.sum).collect Array((tom,CompactBuffer(1, 8, 2)), (jerry,CompactBuffer(9, 2)), (shuke,CompactBuffer(7)), (kitty,CompactBuffer(3)))

WordCount

sc.textFile("/root/words.txt").flatMap(x=>x.split(" ")).map((,1)).reduceByKey(+).sortBy(.2,false).collect sc.textFile("/root/words.txt").flatMap(x=>x.split(" ")).map((,1)).groupByKey.map(t=>(t._1, t._2.sum)).collect

cogroup

val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2))) val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2))) val rdd3 = rdd1.cogroup(rdd2) val rdd4 = rdd3.map(t=>(t._1, t._2._1.sum + t._2._2.sum))

cartesian笛卡尔积

val rdd1 = sc.parallelize(List("tom", "jerry")) val rdd2 = sc.parallelize(List("tom", "kitty", "shuke")) val rdd3 = rdd1.cartesian(rdd2)

###################################################################################################

spark action

val rdd1 = sc.parallelize(List(1,2,3,4,5), 2)

collect

rdd1.collect

reduce

val r = rdd1.reduce(+)

count

rdd1.count

top

rdd1.top(2)

take

rdd1.take(2)

first(similer to take(1))

rdd1.first

takeOrdered

rdd1.takeOrdered(3)

http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html

mapPartitionsWithIndex val func = (index: Int, iter: Iterator[(String)]) => { iter.map(x => "[partID:" + index + ", val: " + x + "]") }

mapPartitionsWithIndex val func = (index: Int, iter: Iterator[Int]) => { iter.map(x => "[partID:" + index + ", val: " + x + "]") } val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2) rdd1.mapPartitionsWithIndex(func).collect



aggregate

def func1(index: Int, iter: Iterator[(Int)]) : Iterator[String] = { iter.toList.map(x => "[partID:" + index + ", val: " + x + "]").iterator } val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2) rdd1.mapPartitionsWithIndex(func1).collect rdd1.aggregate(0)(math.max(_, _), _ + ) rdd1.aggregate(5)(math.max(, _), _ + _)

val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2) def func2(index: Int, iter: Iterator[(String)]) : Iterator[String] = { iter.toList.map(x => "[partID:" + index + ", val: " + x + "]").iterator } rdd2.aggregate("")(_ + _, _ + ) rdd2.aggregate("=")( + _, _ + _)

val rdd3 = sc.parallelize(List("12","23","345","4567"),2) rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)

val rdd4 = sc.parallelize(List("12","23","345",""),2) rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)

val rdd5 = sc.parallelize(List("12","23","","345"),2) rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)



aggregateByKey

val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2) def func2(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = { iter.map(x => "[partID:" + index + ", val: " + x + "]") } pairRDD.mapPartitionsWithIndex(func2).collect pairRDD.aggregateByKey(0)(math.max(_, _), _ + ).collect pairRDD.aggregateByKey(100)(math.max(, _), _ + _).collect



checkpoint sc.setCheckpointDir("hdfs://node-1.edu360.cn:9000/ck") val rdd = sc.textFile("hdfs://node-1.edu360.cn:9000/wc").flatMap(.split(" ")).map((, 1)).reduceByKey(+) rdd.checkpoint rdd.isCheckpointed rdd.count rdd.isCheckpointed rdd.getCheckpointFile



coalesce, repartition val rdd1 = sc.parallelize(1 to 10, 10) val rdd2 = rdd1.coalesce(2, false) rdd2.partitions.length



collectAsMap val rdd = sc.parallelize(List(("a", 1), ("b", 2))) rdd.collectAsMap



combineByKey val rdd1 = sc.textFile("hdfs://node-1.edu360.cn:9000/wc").flatMap(.split(" ")).map((, 1)) val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n) rdd2.collect

val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n) rdd3.collect

val rdd4 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3) val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3) val rdd6 = rdd5.zip(rdd4) val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)



countByKey

val rdd1 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1))) rdd1.countByKey rdd1.countByValue



filterByRange

val rdd1 = sc.parallelize(List(("e", 5), ("c", 3), ("d", 4), ("c", 2), ("a", 1))) val rdd2 = rdd1.filterByRange("b", "d") rdd2.colllect



flatMapValues val a = sc.parallelize(List(("a", "1 2"), ("b", "3 4"))) rdd3.flatMapValues(_.split(" "))



foldByKey

val rdd1 = sc.parallelize(List("dog", "wolf", "cat", "bear"), 2) val rdd2 = rdd1.map(x => (x.length, x)) val rdd3 = rdd2.foldByKey("")(+)

val rdd = sc.textFile("hdfs://node-1.edu360.cn:9000/wc").flatMap(.split(" ")).map((, 1)) rdd.foldByKey(0)(+)



foreachPartition val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3) rdd1.foreachPartition(x => println(x.reduce(_ + _)))



keyBy val rdd1 = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3) val rdd2 = rdd1.keyBy(_.length) rdd2.collect



keys values val rdd1 = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2) val rdd2 = rdd1.map(x => (x.length, x)) rdd2.keys.collect rdd2.values.collect



mapPartitions( it: Iterator => {it.map(x => x * 10)})

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏码匠的流水账

聊聊flink的BoltWrapper

flink-storm_2.11-1.6.2-sources.jar!/org/apache/flink/storm/wrappers/BoltWrapper....

1042
来自专栏HansBug's Lab

2431: [HAOI2009]逆序对数列

2431: [HAOI2009]逆序对数列 Time Limit: 5 Sec  Memory Limit: 128 MB Submit: 954  Solv...

2896
来自专栏10km的专栏

java:java.util.Map和java.util.Set的Key类型转换

google的guava项目提对Map供了丰富强大的转换功能(参见guava jar包中的com.google.common.collect.Maps ) ...

2385
来自专栏数据之美

Spark函数讲解: combineByKey

1、背景 在数据分析中,处理Key,Value的Pair数据是极为常见的场景,例如我们可以针对这样的数据进行分组、聚合或者将两个包含Pair数据的RDD根据ke...

2516
来自专栏desperate633

Java Generic 自定义泛型如何自定义泛型自定义泛型的边界共变性,逆变性泛型对象的比较

考虑我们要实现了一个节点对象,这个对象可以自定义类型,我们可以用泛型语法进行如下的定义:

771
来自专栏闵开慧

spark使用zipWithIndex和zipWithUniqueId为rdd中每条数据添加索引数据

spark的rdd中数据需要添加自增主键,然后将数据存入数据库,使用map来添加有的情况是可以的,有的情况是不可以的,所以需要使用以下两种中的其中一种来进行添加...

7659
来自专栏Spark学习技巧

Java transient关键字使用小记

1. transient的作用及使用方法 我们都知道一个对象只要实现了Serilizable接口,这个对象就可以被序列化,java的这种序列化模式为开发者提供了...

3086
来自专栏加米谷大数据

Spark RDD Map Reduce 基本操作

RDD是Spark中的抽象数据结构类型,任何数据在Spark中都被表示为RDD。从编程的角度来看,RDD可以简单看成是一个数组。和普通数组的区别是,RDD中的数...

1502
来自专栏个人分享

spark RDD transformation与action函数整理

3.过滤 filter 需要注意的是 filter并不会在原有RDD上过滤,而是根据filter的内容重新创建了一个RDD

2092
来自专栏前端儿

JavaScript 面向对象继承详解

由于js不像java那样是完全面向对象的语言,js是基于对象的,它没有类的概念。所以,要想实现继承,一般都是基于原型链的方式;

661

扫码关注云+社区

领取腾讯云代金券