Spark学习笔记——键值对操作
2017-04-07 18:50
435 查看
键值对 RDD是 Spark 中许多操作所需要的常见数据类型
键值对 RDD 通常用来进行聚合计算。我们一般要先通过一些初始 ETL(抽取、转化、装载)操作来将数据转化为键值对形式。
Spark 为包含键值对类型的 RDD 提供了一些专有的操作。
1.创建Pair RDD
2.Pair RDD的转化操作
Pair RDD 可以使用所有标准 RDD 上的可用的转化操作。
Pair RDD也支持RDD所支持的函数
3.聚合操作
类似fold() 、 combine() 、 reduce() 等行动操作,这些操作返回 RDD,因此它们是转化操作而不是行动操作。
reduceByKey()函数,接收一个函数,并使用这个函数对值进行合并
或者
foldByKey()函数,也是接收一个函数,并使用这个函数对值进行合并,提供初始值
countByValue()函数
combineByKey()函数,使用 combineByKey() 求每个键对应的平均值
4.数据分组
groupByKey()函数,将数据根据键进行分组
5.连接
join操作符
有时,我们不希望结果中的键必须在两个 RDD 中都存在,join的部分可以不存在
leftOuterJoin(other) 和 rightOuterJoin(other) 都会根据键连接两个 RDD,但是允许结果中存在其中的一个 pair RDD 所 缺失的键。
leftOuterJoin(other)结果
rightOuterJoin(other)结果
6.数据排序
在 Scala 中以字符串顺序对整数进行自定义排序
键值对 RDD 通常用来进行聚合计算。我们一般要先通过一些初始 ETL(抽取、转化、装载)操作来将数据转化为键值对形式。
Spark 为包含键值对类型的 RDD 提供了一些专有的操作。
1.创建Pair RDD
val input = sc.parallelize(List(1, 2, 3, 4)) val pairs = input.map(x => (x+1, x)) for (pair <- pairs){ println(pair) } //输出 (2,1) (3,2) (4,3) (5,4)
2.Pair RDD的转化操作
Pair RDD 可以使用所有标准 RDD 上的可用的转化操作。
Pair RDD也支持RDD所支持的函数
pairs.filter{case (key, value) => value.length < 20}
3.聚合操作
类似fold() 、 combine() 、 reduce() 等行动操作,这些操作返回 RDD,因此它们是转化操作而不是行动操作。
reduceByKey()函数,接收一个函数,并使用这个函数对值进行合并
val wordCount = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b) #切分成单词,转换成键值对并计数
或者
input.flatMap(x => x.split(" ")).countByValue()
foldByKey()函数,也是接收一个函数,并使用这个函数对值进行合并,提供初始值
rdd.mapValues(x => (x, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
countByValue()函数
val textFile = sc.textFile(inputFile) val result1 = textFile.flatMap(x => x.split(" ")).countByValue() println(result1) 输出:Map(cc -> 3, aa -> 6, bb -> 3, ee -> 3, ff -> 2, hh -> 1, dd -> 1, gg -> 1)
combineByKey()函数,使用 combineByKey() 求每个键对应的平均值
val data = Seq(("a", 3), ("b", 4), ("a", 1)) val input2 = sc.parallelize(data) //使用 combineByKey() 求每个键对应的平均值 val result2 = input2.combineByKey( (v) => (v, 1), (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1), (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2) ).map{ case (key, value) => (key, value._1 / value._2.toFloat) } result2.collectAsMap().map(println(_)) //输出 (b,4.0) (a,2.0)
4.数据分组
groupByKey()函数,将数据根据键进行分组
val data = Seq(("a", 3), ("b", 4), ("a", 1)) val input2 = sc.parallelize(data) val result3 = input2.groupByKey() for (result <- result3) println(result) //输出: (a,CompactBuffer(3, 1)) (b,CompactBuffer(4))
5.连接
join操作符
val input = sc.parallelize(List(1, 2, 3, 4)) val pairs1 = input.map(x => (x+1, x)) //输出(2,1),(3,2),(4,3),(5,4) val pairs2 = input.map(x => (x+1, 1)) //输出(2,1),(3,1),(4,1),(5,1) for(pair <- pairs1.join(pairs2)){ println(pair) } //输出(4,(3,1)),(3,(2,1)),(5,(4,1)),(2,(1,1))
有时,我们不希望结果中的键必须在两个 RDD 中都存在,join的部分可以不存在
leftOuterJoin(other) 和 rightOuterJoin(other) 都会根据键连接两个 RDD,但是允许结果中存在其中的一个 pair RDD 所 缺失的键。
leftOuterJoin(other)结果
(4,(3,Some(1))) (3,(2,Some(1))) (5,(4,Some(1))) (2,(1,Some(1)))
rightOuterJoin(other)结果
(4,(Some(3),1)) (3,(Some(2),1)) (5,(Some(4),1)) (2,(Some(1),1))
6.数据排序
在 Scala 中以字符串顺序对整数进行自定义排序
val input = sc.parallelize(List(1, 2, 3, 4)) val pairs1 = input.map(x => (x + 1, x)) implicit val sortIntegersByString = new Ordering[Int] { override def compare(a: Int, b: Int) = b.toString.compare(a.toString) } for(pair <- pairs1.sortByKey()) println(pair) //输出 (5,4) (4,3) (3,2) (2,1)
相关文章推荐
- Spark学习笔记-键值对操作
- Spark学习之键值对(pair RDD)操作(3)
- Spark学习之键值对(pair RDD)操作
- Spark学习之键值对(pair RDD)操作(3)
- Spark Hadoop集群部署与Spark操作HDFS运行详解---Spark学习笔记10
- spark学习笔记:flatMap()(API操作报错)
- Spark学习笔记-HBase操作
- Spark的transformation 和 action的操作学习笔记
- Spark学习笔记(26)在DStream的Action操作之外也可能产生Job操作
- Spark Hadoop集群部署与Spark操作HDFS运行详解---Spark学习笔记10
- Spark学习之四-键值对操作
- spark学习笔记:集群模式下的addFile()操作(存疑)
- scala学习笔记5 spark-shell的简单操作
- Scala中隐式转换内幕操作规则揭秘、最佳实践及其在Spark中的应用源码解析之Scala学习笔记-55
- Linq学习笔记2(查询操作)
- [mfc学习笔记一]以一个CGdiObject类的绘图例子来看一下绘图操作
- 连接操作 - Oracle 学习笔记 3
- C#学习笔记一(字符串操作)
- LINQ学习笔记三-----基本查询操作
- 孙鑫VC学习笔记:第十二讲 (四) 用API函数、CFile类操作文件