Spark学习笔记3:键值对操作

时间:2022-10-22 21:56:58

键值对RDD通常用来进行聚合计算,Spark为包含键值对类型的RDD提供了一些专有的操作。这些RDD被称为pair RDD。pair RDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口。

Spark中创建pair RDD的方法:存储键值对的数据格式会在读取时直接返回由其键值对数据组成的pair RDD,还可以使用map()函数将一个普通的RDD转为pair RDD。

  • Pair RDD的转化操作
  1. reduceByKey()  与reduce类似 ,接收一个函数,并使用该函数对值进行合并,为每个数据集中的每个键进行并行的归约操作。返回一个由各键和对应键归约出来的结果值组成的新的RDD。例如 :上一章中单词计数的例子:val counts  =  words.map(word => (word,1)).reduceByKey{ case (x,y) => x + y}
  2. foldByKey()与fold()类似,都使用一个与RDD和合并函数中的数据类型相同的零值最为初始值。val counts  =  words.map(word => (word,1)).foldByKey{ case (x,y) => x + y}
  3. combineByKey()是最为常用的基于键进行聚合的函数,可以返回与输入类型不同的返回值。

  理解combineByKey处理数据流程,首先需要知道combineByKey的createCombiner()函数用来创建那个键对应的累加器的初始值,mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并。mergeCombiners()方法将各个分区的结果进行合并。

使用combineByKey进行单词计数的例子:

import org.apache.spark.{SparkConf, SparkContext}

object word {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("word")
val sc = new SparkContext(conf)
val input = sc.parallelize(List(("coffee",1),("coffee",2),("panda",3),("coffee",9)))
val counts = input.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)
)
counts.foreach(println)
}
}

输出结果:

Spark学习笔记3:键值对操作

这个例子中的数据流示意图如下:

Spark学习笔记3:键值对操作

简单说过程就是,将输入键值对数据进行分区,每个分区先根据键计算相应的值以及键出现的次数。然后对不同分区进行合并得出最后的结果。

  4.groupByKey()使用RDD中的键来对数据进行分组,对于一个由类型K的键和类型V的值组成的RDD,所得到的结果RDD类型会是[K, Iterable[V] ]

 例如:

import org.apache.spark.{SparkConf, SparkContext}

object word {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("word")
val sc = new SparkContext(conf)
val input = sc.parallelize(List("scala spark scala core scala python java spark scala"))
val words = input.flatMap(line => line.split(" ")).map(word => (word,1))
val counts = words.groupByKey()
counts.foreach(println)
}
}

  输出:

Spark学习笔记3:键值对操作

  5、cogroup函数对多个共享同一个键的RDD进行分组,对两个键类型均为K而值类型分别为V和W的RDD进行cogroup时,得到的结果RDD类型为[(K,(Iterable[V],Iterable[W]))]

  6、join(other)这样的连接是内连接,只有在两个pair RDD中都存在的键才输出。若一个输入对应的键有多个值时,生成的pair RDD会包括来自两个输入RDD的每一组相对应的记录。理解这句话看下面的例子:

val rdd = sc.parallelize(List((1,2),(3,4),(3,6)))
val other = sc.parallelize(List((3,9)))
val joins = rdd.join(other)

  输出结果:

Spark学习笔记3:键值对操作

  7、leftOuterJoin(other)左外连接和rightOuterJoin(other)右外连接都会根据键连接两个RDD,但是允许结果中存在其中的一个pair RDD所缺失的键。

val rdd = sc.parallelize(List((1,2),(3,4),(3,6)))
val other = sc.parallelize(List((3,9)))
val join1 = rdd.rightOuterJoin(other)

  输出结果:

Spark学习笔记3:键值对操作

val rdd = sc.parallelize(List((1,2),(3,4),(3,6)))
val other = sc.parallelize(List((3,9)))
val join2 = rdd.leftOuterJoin(other)

  输出结果: 

 Spark学习笔记3:键值对操作

  8、sortByKey()函数接收一个叫做ascending的参数,表示想要让结果升序排序还是降序排序。

val input = sc.parallelize(List("scala spark scala core scala python java spark scala"))
val words = input.flatMap(line => line.split(" ")).map(word => (word,1)).reduceByKey((x,y)=>x+y)
val counts = words.sortByKey()

  输出结果:

Spark学习笔记3:键值对操作

  •  Pair RDD的行动操作
  1. countByKey() 对每个键对应的元素分别计数。
  2. collectAsMap()将结果以映射表的形式返回,注意后面的value会覆盖前面的。
    val num = sc.parallelize(List((1,2),(3,4),(3,6)))
    println(num.collectAsMap().mkString(" "))

    输出结果:Spark学习笔记3:键值对操作

  3. lookup(key)返回给定键对应的所有值。
  • 数据分区

  Spark程序可以通过控制RDD分区方式来减少通信开销。

运行下面这段代码,用来查看用户查阅了自己订阅的主题的页面的数量,结果返回3:

val list1 =List(Tuple2("Mike",List("sports","math")),Tuple2("Jack",List("travel","book")))//UserID用户ID,UserInfo用户订阅的主题
val list2= List(Tuple2("Mike","sports"),Tuple2("Mike","stock"),Tuple2("Jack","travel"),Tuple2("Jack","book"))//UserID,LinkInfo用户访问情况
val userData = sc.parallelize(list1)
val events = sc.parallelize(list2)
userData.persist()
val joined = userData.join(events)
val results = joined.filter({
case (id, (info, link)) =>
info.contains(link)
}
).count()
println(results)

  上面这段代码中,用到了join操作,会将两个数据集中的所有键的哈希值都求出来,将该哈希值相同的记录通过网络传到同一台机器上,然后在那台机器上对所有键相同的记录进行连接操作。

  假如userdata表很大很大,而且几乎是不怎么变化的,那么每次都对userdata表进行哈希值计算和跨节点的数据混洗,就会产生很多的额外开销。

如下:

Spark学习笔记3:键值对操作

解决这一产生额外开销的方法就是,对userdata表使用partitionBy()转化操作,将这张表转为哈希分区。修改后的代码如下:

    val list1 =List(Tuple2("Mike",List("sports","math")),Tuple2("Jack",List("travel","book")))//UserID用户ID,UserInfo用户订阅的主题
val list2= List(Tuple2("Mike","sports"),Tuple2("Mike","stock"),Tuple2("Jack","travel"),Tuple2("Jack","book"))//UserID,LinkInfo用户访问情况
val userData = sc.parallelize(list1)
val events = sc.parallelize(list2)
userData.partitionBy(new DomainNamePartitioner(10)).persist()
val joined = userData.join(events)
val results = joined.filter({
case (id, (info, link)) =>
info.contains(link)
}
).count()
println(results)

  构建userData时调用了partitionBy(),在调用join()时,Spark只会对events进行数据混洗操作,将events中特定UserID的记录发送到userData的对应分区所在的那台机器上。这样,通过网络传输的数据就大大减少,程序运行速度也可以显著提升。partitionBy()是一个转化操作,因此它的返回值是一个新的RDD。

  新的数据处理过程如下:

Spark学习笔记3:键值对操作

  scala可以使用RDD的partitioner属性来获取RDD的分区方式,它会返回一个scala.Option对象。

  可以从数据分区中获益的操作有cogroup() , groupWith() , join() , leftOuterJoin() , rightOuterJoin() , groupByKey() , reduceByKey() , combineByKey()以及lookup()。

  实现自定义分区器,需要继承org.apache.spark.Partitioner类并实现下面的三个方法:

  • numPartitions: Int :返回创建出来的分区数
  • getPartition(key: Any):Int : 返回给定键的分区编号(0 到 numPartitions - 1)
  • equals() : Java判断相等的方法,Spark用这个方法来检查分区器对象是否和其他分区器实例相同,这样Spark才可以判断两个RDD的分区方式是否相同。

Spark学习笔记3:键值对操作的更多相关文章

  1. 【原】Learning Spark (Python版) 学习笔记(二)----键值对、数据读取与保存、共享特性

    本来应该上周更新的,结果碰上五一,懒癌发作,就推迟了 = =.以后还是要按时完成任务.废话不多说,第四章-第六章主要讲了三个内容:键值对.数据读取与保存与Spark的两个共享特性(累加器和广播变量). ...

  2. Spark学习之键值对操作总结

    键值对 RDD 是 Spark 中许多操作所需要的常见数据类型.键值对 RDD 通常用来进行聚合计算.我们一般要先通过一些初始 ETL(抽取.转化.装载)操作来将数据转化为键值对形式.键值对 RDD ...

  3. Spark中的键值对操作-scala

    1.PairRDD介绍     Spark为包含键值对类型的RDD提供了一些专有的操作.这些RDD被称为PairRDD.PairRDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口.例如,Pa ...

  4. Spark中的键值对操作

    1.PairRDD介绍     Spark为包含键值对类型的RDD提供了一些专有的操作.这些RDD被称为PairRDD.PairRDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口.例如,Pa ...

  5. Spark学习笔记之SparkRDD

    Spark学习笔记之SparkRDD 一.   基本概念 RDD(resilient distributed datasets)弹性分布式数据集. 来自于两方面 ①   内存集合和外部存储系统 ②   ...

  6. Spark学习笔记3——RDD(下)

    目录 Spark学习笔记3--RDD(下) 向Spark传递函数 通过匿名内部类 通过具名类传递 通过带参数的 Java 函数类传递 通过 lambda 表达式传递(仅限于 Java 8 及以上) 常 ...

  7. Spark学习笔记1——第一个Spark程序:单词数统计

    Spark学习笔记1--第一个Spark程序:单词数统计 笔记摘抄自 [美] Holden Karau 等著的<Spark快速大数据分析> 添加依赖 通过 Maven 添加 Spark-c ...

  8. spark学习笔记总结-spark入门资料精化

    Spark学习笔记 Spark简介 spark 可以很容易和yarn结合,直接调用HDFS.Hbase上面的数据,和hadoop结合.配置很容易. spark发展迅猛,框架比hadoop更加灵活实用. ...

  9. X-Cart 学习笔记(四)常见操作

    目录 X-Cart 学习笔记(一)了解和安装X-Cart X-Cart 学习笔记(二)X-Cart框架1 X-Cart 学习笔记(三)X-Cart框架2 X-Cart 学习笔记(四)常见操作 五.常见 ...

  10. Spark学习笔记2——RDD(上)

    目录 Spark学习笔记2--RDD(上) RDD是什么? 例子 创建 RDD 并行化方式 读取外部数据集方式 RDD 操作 转化操作 行动操作 惰性求值 Spark学习笔记2--RDD(上) 笔记摘 ...

随机推荐

  1. 如何写出让java虚拟机发生内存溢出异常OutOfMemoryError的代码

    程序小白在写代码的过程中,经常会不经意间写出发生内存溢出异常的代码.很多时候这类异常如何产生的都傻傻弄不清楚,如果能故意写出让jvm发生内存溢出的代码,有时候看来也并非一件容易的事.最近通过学习&lt ...

  2. 《Entity Framework 6 Recipes》中文翻译系列 &lpar;6&rpar; -----第二章 实体数据建模基础之使用Code First建模自引用关系

    2-5 使用Code First建模自引用关系 问题 你的数据库中一张自引用的表,你想使用Code First 将其建模成一个包含自关联的实体. 解决方案 我们假设你有如图2-14所示的数据库关系图的 ...

  3. CodeForces152C——Pocket Book&lpar;排列组合问题&rpar;

    Pocket Book DescriptionOne day little Vasya found mom's pocket book. The book had n names of her fri ...

  4. POJ 2391 容牛问题

    题目大意:给定一个无向图,点i处有Ai头牛,点i处的牛棚能容纳Bi头牛,求一个最短时间T使得在T时间内所有的牛都能进到某一牛棚里去.(1 <= N <= 200, 1 <= M &l ...

  5. android 实现与服务器的长链接 方式

    http://blog.csdn.net/coffeeco/article/details/13276437  这边文章主要看服务端,使用tomcat7以上实现服务端的接收消息以及消息发送 http: ...

  6. 异常驱动的开发&lpar;Exception-Driven Development&rpar;

    你的网站或应用程序存在哪些问题?如果你在等着用户来告诉你,那么你只能看到所有的问题中已经暴露的那极小的一部分.要知道,那只是"冰山一角"! 而且,如果你真的是在守株待兔,我不得不很 ...

  7. 第三章 启动rabbitmq的webUI

    一.启动步骤 1.启动rabbitmq rabbitmq-server (前台启动)或者rabbitmq-server -detached(后台启动) 2.启动rabbitmq_management ...

  8. ios-键盘回收

    对输入的UITextField 控件设置完代理,然后粘上两种方法,即可. #pragma mark- 键盘收回 -(void)textFieldDidEndEditing:(UITextField * ...

  9. MVC 深入讲解Routing &lowbar;路由规则【八】

    一.客户端=>控制器 在项目中我们引用了system.web.routing, 如果第一个匹配成功了,那么后面的都不会再匹配. 1. routing的作用: 确定colltroller,确定ac ...

  10. django url 中的namespace详解

    [引子] django是一个比较全能的web框架,比如它实现了自己的orm模型来对接不同的数据库.并提供了方便的数据接口:django牛逼的地方 还有许多,如他实现了自己的模板语言,这样我们就能方便的 ...