Spark之键值RDD转换(转载)

1.mapValus(fun):对[K,V]型数据中的V值map操作
(例1)：对每个的的年龄加2

object MapValues {
  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local").setAppName("map")
    val sc = new SparkContext(conf)
    val list = List(("mobin",22),("kpop",20),("lufei",23))
    val rdd = sc.parallelize(list)
    val mapValuesRDD = rdd.mapValues(_+2)
    mapValuesRDD.foreach(println)
  }
}

输出：
(mobin,24)
(kpop,22)
(lufei,25)
(RDD依赖图：红色块表示一个RDD区，黑色块表示该分区集合，下同)

2.flatMapValues(fun)：对[K,V]型数据中的V值flatmap操作
(例2):

//省略<br>val list = List(("mobin",22),("kpop",20),("lufei",23))
val rdd = sc.parallelize(list)
val mapValuesRDD = rdd.flatMapValues(x => Seq(x,"male"))
mapValuesRDD.foreach(println)

输出：
(mobin,22)
(mobin,male)
(kpop,20)
(kpop,male)
(lufei,23)
(lufei,male)
如果是mapValues会输出：
(mobin,List(22, male))
(kpop,List(20, male))
(lufei,List(23, male))
（RDD依赖图）

3.comineByKey(createCombiner,mergeValue,mergeCombiners,partitioner,mapSideCombine)
 
   comineByKey(createCombiner,mergeValue,mergeCombiners,numPartitions)
 
   comineByKey(createCombiner,mergeValue,mergeCombiners)
 
createCombiner:在第一次遇到Key时创建组合器函数，将RDD数据集中的V类型值转换C类型值（V => C），
如例3:

mergeValue：合并值函数，再次遇到相同的Key时，将createCombiner道理的C类型值与这次传入的V类型值合并成一个C类型值（C,V）=>C，
如例3：

mergeCombiners:合并组合器函数，将C类型值两两合并成一个C类型值
如例3：

partitioner：使用已有的或自定义的分区函数，默认是HashPartitioner
 
mapSideCombine：是否在map端进行Combine操作,默认为true
 
注意前三个函数的参数类型要对应；第一次遇到Key时调用createCombiner，再次遇到相同的Key时调用mergeValue合并值
 
（例3）：统计男性和女生的个数，并以（性别，（名字，名字....），个数）的形式输出

object CombineByKey {
  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local").setAppName("combinByKey")
    val sc = new SparkContext(conf)
    val people = List(("male", "Mobin"), ("male", "Kpop"), ("female", "Lucy"), ("male", "Lufei"), ("female", "Amy"))
    val rdd = sc.parallelize(people)
    val combinByKeyRDD = rdd.combineByKey(
      (x: String) => (List(x), 1),
      (peo: (List[String], Int), x : String) => (x :: peo._1, peo._2 + 1),
      (sex1: (List[String], Int), sex2: (List[String], Int)) => (sex1._1 ::: sex2._1, sex1._2 + sex2._2))
    combinByKeyRDD.foreach(println)
    sc.stop()
  }
}

输出：
(male,(List(Lufei, Kpop, Mobin),3))
(female,(List(Amy, Lucy),2))
过程分解：

Partition1:
K="male"  -->  ("male","Mobin")  --> createCombiner("Mobin") =>  peo1 = (  List("Mobin") , 1 )
K="male"  -->  ("male","Kpop")  --> mergeValue(peo1,"Kpop") =>  peo2 = (  "Kpop"  ::  peo1_1 , 1 + 1 )    //Key相同调用mergeValue函数对值进行合并
K="female"  -->  ("female","Lucy")  --> createCombiner("Lucy") =>  peo3 = (  List("Lucy") , 1 )
 
Partition2:
K="male"  -->  ("male","Lufei")  --> createCombiner("Lufei") =>  peo4 = (  List("Lufei") , 1 )
K="female"  -->  ("female","Amy")  --> createCombiner("Amy") =>  peo5 = (  List("Amy") , 1 )
 
Merger Partition:
K="male" --> mergeCombiners(peo2,peo4) => (List(Lufei,Kpop,Mobin))
K="female" --> mergeCombiners(peo3,peo5) => (List(Amy,Lucy))

（RDD依赖图）

4.foldByKey(zeroValue)(func)
 
  foldByKey(zeroValue,partitioner)(func)
 
  foldByKey(zeroValue,numPartitiones)(func)
 
foldByKey函数是通过调用CombineByKey函数实现的
 
zeroVale：对V进行初始化，实际上是通过CombineByKey的createCombiner实现的  V =>  (zeroValue,V)，再通过func函数映射成新的值，即func(zeroValue,V),如例4可看作对每个V先进行  V=> 2 + V  
 
func: Value将通过func函数按Key值进行合并（实际上是通过CombineByKey的mergeValue，mergeCombiners函数实现的，只不过在这里，这两个函数是相同的）
例4：
//省略

    val people = List(("Mobin", 2), ("Mobin", 1), ("Lucy", 2), ("Amy", 1), ("Lucy", 3))
    val rdd = sc.parallelize(people)
    val foldByKeyRDD = rdd.foldByKey(2)(_+_)
    foldByKeyRDD.foreach(println)

输出：
(Amy,2)
(Mobin,4)
(Lucy,6)
先对每个V都加2，再对相同Key的value值相加。
 
 
5.reduceByKey(func,numPartitions):按Key进行分组，使用给定的func函数聚合value值, numPartitions设置分区数，提高作业并行度
例5

//省略

val arr = List(("A",3),("A",2),("B",1),("B",3))
val rdd = sc.parallelize(arr)
val reduceByKeyRDD = rdd.reduceByKey(_ +_)
reduceByKeyRDD.foreach(println)
sc.stop

输出：
(A,5)
(A,4)
（RDD依赖图）

6.groupByKey(numPartitions):按Key进行分组，返回[K,Iterable[V]]，numPartitions设置分区数，提高作业并行度
例6：

//省略

val arr = List(("A",1),("B",2),("A",2),("B",3))
val rdd = sc.parallelize(arr)
val groupByKeyRDD = rdd.groupByKey()
groupByKeyRDD.foreach(println)
sc.stop

输出：
(B,CompactBuffer(2, 3))
(A,CompactBuffer(1, 2))
 
以上foldByKey,reduceByKey,groupByKey函数最终都是通过调用combineByKey函数实现的
 
7.sortByKey(accending，numPartitions):返回以Key排序的（K,V）键值对组成的RDD，accending为true时表示升序，为false时表示降序，numPartitions设置分区数，提高作业并行度
例7：

//省略sc

val arr = List(("A",1),("B",2),("A",2),("B",3))
val rdd = sc.parallelize(arr)
val sortByKeyRDD = rdd.sortByKey()
sortByKeyRDD.foreach(println)
sc.stop

输出：
(A,1)
(A,2)
(B,2)
(B,3)
 
8.cogroup(otherDataSet，numPartitions)：对两个RDD(如:(K,V)和(K,W))相同Key的元素先分别做聚合，最后返回(K,Iterator<V>,Iterator<W>)形式的RDD,numPartitions设置分区数，提高作业并行度
例8：

//省略

val arr = List(("A", 1), ("B", 2), ("A", 2), ("B", 3))
val arr1 = List(("A", "A1"), ("B", "B1"), ("A", "A2"), ("B", "B2"))
val rdd1 = sc.parallelize(arr, 3)
val rdd2 = sc.parallelize(arr1, 3)
val groupByKeyRDD = rdd1.cogroup(rdd2)
groupByKeyRDD.foreach(println)
sc.stop

输出：
(B,(CompactBuffer(2, 3),CompactBuffer(B1, B2)))
(A,(CompactBuffer(1, 2),CompactBuffer(A1, A2)))
（RDD依赖图）

9.join(otherDataSet,numPartitions):对两个RDD先进行cogroup操作形成新的RDD，再对每个Key下的元素进行笛卡尔积，numPartitions设置分区数，提高作业并行度
例9
//省略

val arr = List(("A", 1), ("B", 2), ("A", 2), ("B", 3))
val arr1 = List(("A", "A1"), ("B", "B1"), ("A", "A2"), ("B", "B2"))
val rdd = sc.parallelize(arr, 3)
val rdd1 = sc.parallelize(arr1, 3)
val groupByKeyRDD = rdd.join(rdd1)
groupByKeyRDD.foreach(println)

输出：

(B,(2,B1))
(B,(2,B2))
(B,(3,B1))
(B,(3,B2))
 
(A,(1,A1))
(A,(1,A2))
(A,(2,A1))
(A,(2,A2)

（RDD依赖图）

10.LeftOutJoin(otherDataSet，numPartitions):左外连接，包含左RDD的所有数据，如果右边没有与之匹配的用None表示,numPartitions设置分区数，提高作业并行度
例10：
//省略

val arr = List(("A", 1), ("B", 2), ("A", 2), ("B", 3),("C",1))
val arr1 = List(("A", "A1"), ("B", "B1"), ("A", "A2"), ("B", "B2"))
val rdd = sc.parallelize(arr, 3)
val rdd1 = sc.parallelize(arr1, 3)
val leftOutJoinRDD = rdd.leftOuterJoin(rdd1)
leftOutJoinRDD .foreach(println)
sc.stop

输出：

(B,(2,Some(B1)))
(B,(2,Some(B2)))
(B,(3,Some(B1)))
(B,(3,Some(B2)))
 
(C,(1,None))
 
(A,(1,Some(A1)))
(A,(1,Some(A2)))
(A,(2,Some(A1)))
(A,(2,Some(A2)))

11.RightOutJoin(otherDataSet, numPartitions):右外连接，包含右RDD的所有数据，如果左边没有与之匹配的用None表示,numPartitions设置分区数，提高作业并行度
例11：
//省略

val arr = List(("A", 1), ("B", 2), ("A", 2), ("B", 3))
val arr1 = List(("A", "A1"), ("B", "B1"), ("A", "A2"), ("B", "B2"),("C","C1"))
val rdd = sc.parallelize(arr, 3)
val rdd1 = sc.parallelize(arr1, 3)
val rightOutJoinRDD = rdd.rightOuterJoin(rdd1)
rightOutJoinRDD.foreach(println)
sc.stop

输出：

(B,(Some(2),B1))
(B,(Some(2),B2))
(B,(Some(3),B1))
(B,(Some(3),B2))
 
(C,(None,C1))
 
(A,(Some(1),A1))
(A,(Some(1),A2))
(A,(Some(2),A1))
(A,(Some(2),A2))

顶
0
踩
时间： 2024-11-03 03:34:26

Spark之键值RDD转换(转载)的相关文章

Spark常用函数讲解--键值RDD转换

摘要: RDD:弹性分布式数据集,是一种特殊集合 ‚ 支持多种来源 ‚ 有容错机制 ‚ 可以被缓存 ‚ 支持并行操作,一个RDD代表一个分区里的数据集RDD有两种操作算子:         Transformation(转换):Transformation属于延迟计算,当一个RDD转换成另一个RDD时并没有立即进行转换,仅仅是记住       了数据集的逻辑操作         Ation(执行):触发Spark作业的运行,真正触发转换算子的计算 本系列主要讲解Spark中常用的函数操作:   

Spark笔记：复杂RDD的API的理解（上）

本篇接着讲解RDD的API,讲解那些不是很容易理解的API,同时本篇文章还将展示如何将外部的函数引入到RDD的API里使用,最后通过对RDD的API深入学习,我们还讲讲一些和RDD开发相关的scala语法. 1)  aggregate(zeroValue)(seqOp,combOp)  该函数的功能和reduce函数一样,也是对数据进行聚合操作,不过aggregate可以返回和原RDD不同的数据类型,使用时候还要提供初始值. 我们来看看下面的用法,代码如下: val rddInt: RDD[In

大数据技术之_19_Spark学习_02_Spark Core 应用解析+ RDD 概念 + RDD 编程 + 键值对 RDD + 数据读取与保存主要方式 + RDD 编程进阶 + Spark Core 实例练习

第1章 RDD 概念1.1 RDD 为什么会产生1.2 RDD 概述1.2.1 什么是 RDD1.2.2 RDD 的属性1.3 RDD 弹性1.4 RDD 特点1.4.1 分区1.4.2 只读1.4.3 依赖1.4.4 缓存1.4.5 CheckPoint第2章 RDD 编程2.1 RDD 编程模型2.2 RDD 创建2.2.1 由一个已经存在的 Scala 集合创建,即集合并行化(测试用)2.2.2 由外部存储系统的数据集创建(开发用)2.3 RDD 编程2.3.1 Transformatio

Spark RDD常用算子操作（八） 键值对关联操作 subtractByKey, join,fullOuterJoin, rightOuterJoin, leftOuterJoin

原文作者:翟开顺首发:CSDN本人仅为自己方便查阅做了摘抄,请支持原作者原文地址:https://blog.csdn.net/t1dmzks/article/details/72077428 github: https://github.com/zhaikaishun/spark_tutorial/tree/master/src/main/java/com/spark/rdd_tutorial/tutorial8先从spark-learning中的一张图大致了解其功能 subtractByKey

Spark中的键值对操作-scala

1.PairRDD介绍     Spark为包含键值对类型的RDD提供了一些专有的操作.这些RDD被称为PairRDD.PairRDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口.例如,PairRDD提供了reduceByKey()方法,可以分别规约每个键对应的数据,还有join()方法,可以把两个RDD中键相同的元素组合在一起,合并为一个RDD. 2.创建Pair RDD 程序示例:对一个英语单词组成的文本行,提取其中的第一个单词作为key,将整个句子作为value,建立 PairR

Learning Spark中文版--第四章--使用键值对（2）

Actions Available on Pair RDDs (键值对RDD可用的action) ??和transformation(转换)一样,键值对RDD也可以使用基础RDD上的action(开工),并且键值对RDD有一些利用键值对数据特性的的action,如下表: 表4-3 键值对RDD上的action 函数名 描述 例子 结果 countByKey() 计算每个键元素的总数 rdd.countByKey() {(1,1),(3,2)} collectAsMap() 结果收集成一个map便

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

键值对RDD通常用来进行聚合计算,Spark为包含键值对类型的RDD提供了一些专有的操作.这些RDD被称为pair RDD.pair RDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口. Spark中创建pair RDD的方法:存储键值对的数据格式会在读取时直接返回由其键值对数据组成的pair RDD,还可以使用map()函数将一个普通的RDD转为pair RDD. Pair RDD的转化操作 reduceByKey()  与reduce类似 ,接收一个函数,并使用该函数对值进行合并,

5.2 键值对RDD

一.键值对RDD的创建 1.从文件中加载 2.通过并行集合(数组)创建RDD 二.常用的键值对RDD转换操作 三.综合实例 原文地址:https://www.cnblogs.com/nxf-rabbit75/p/11811930.html

键值转换--复杂按键事件识别程序（链接层）

复杂按键事件识别程序(链接层) 复杂按键包括多个按键的单按事件.长按事件.连按事件.组合按键事件等 本程序为链接层程序,处理和识别按键驱动层扫描到的按键信息g_u8KeyValue.(g_u8KeyValue的每一位表示一个按键) 单按事件在按下键松开时判定,其它按键情况在达到规定的时间后触发相应按键事件. 宏定义: //宏定义各按键按下时的bit值 #define KEY_EVENT_NULL_CLICK 0x00 #define KEY_EVENT_CLOCK_CLICK 0x01 #def