赞
踩
键值对 RDD 就是每个RDD的元素都是 (key,value)类型的键值对,是一种常见的 RDD,可以应用于很多场景。
因为毕竟通过我们之前Hadoop的学习中,我们就可以看到对数据的处理,基本都是以键值对的形式进行统一批处理的,因为MapReduce模型中,Mapper和Reducer之间的联系就是通过键和值进行连接产生关系的。
其实就是个RDD 的创建,无非就是通过并行集合创建和通过文件系统创建,然后文件系统又分为本地文件系统和HDFS。
和上一篇文章中的用法一致。
和上一篇文章中的用法一致。
返回键值对 RDD 中所有的key,构成一个新的 RDD。
- import org.apache.spark.rdd.RDD
- import org.apache.spark.{SparkConf, SparkContext}
-
- object KV_RDD {
-
- def main(args: Array[String]): Unit = {
- //创建SparkContext对象
- val conf = new SparkConf()
- conf.setAppName("kv_rdd").setMaster("local")
- val sc:SparkContext = new SparkContext(conf)
-
- //通过并行集合创建RDD
- val arr = Array(("Spark",1),("Hadoop",1),("Spark",1),("Flink",1))
- val rdd: RDD[(String, Int)] = sc.parallelize(arr)
-
- val res: RDD[String] = rdd.keys
- res.foreach(println)
-
- //关闭SparkContext
- sc.stop()
- }
- }
输出结果:
- Spark
- Hadoop
- Spark
- Flink
返回键值对 RDD 中所有的key,构成一个新的 RDD。
- //通过并行集合创建RDD
- val arr = Array(("Spark",1),("Hadoop",1),("Spark",1),("Flink",1))
- val rdd: RDD[(String, Int)] = sc.parallelize(arr)
-
- val res: RDD[Int] = rdd.values
- res.foreach(println)
运行结果:
- 1
- 1
- 1
- 1
返回一个根据 key 排序(字典序)的RDD。
- //通过并行集合创建RDD
- val arr = Array(("Spark",1),("Hadoop",1),("Spark",1),("Flink",1))
- val rdd: RDD[(String, Int)] = sc.parallelize(arr)
-
- val res: RDD[(String,Int)] = rdd.sortByKey()
- res.foreach(println)
运行结果:
- (Flink,1)
- (Hadoop,1)
- (Spark,1)
- (Spark,1)
默认我们sortByKey()方法是升序排序的,如果要降序可以传入一个false的值。
- //通过并行集合创建RDD
- val arr = Array(("Spark",1),("Hadoop",1),("Spark",1),("Flink",1))
- val rdd: RDD[(String, Int)] = sc.parallelize(arr)
- //降序
- val res: RDD[(String,Int)] = rdd.sortByKey(false)
- res.foreach(println)
运行结果:
- (Spark,1)
- (Spark,1)
- (Hadoop,1)
- (Flink,1)
可以根据其他字段进行排序。
- //通过并行集合创建RDD
- val arr = Array(("Spark",1),("Hadoop",5),("Hive",2),("Flink",3))
- val rdd: RDD[(String, Int)] = sc.parallelize(arr)
-
- //按照value升序排序
- val res: RDD[(String,Int)] = rdd.sortBy(kv=>kv._2,true)
- res.foreach(println)
运行结果:
- (Spark,1)
- (Hive,2)
- (Flink,3)
- (Hadoop,5)
之前我们处理的RDD 都是文本或数字类型的,之前我们的map(func)中的func函数是对整个RDD的元素进行处理。但是这里换成了mapValues(func),这里func函数处理的是我们(key,value)中的所有value,而key 不会发生变化。
- //通过并行集合创建RDD
- val arr = Array(("Spark",1),("Hadoop",5),("Hive",2),("Flink",3))
- val rdd: RDD[(String, Int)] = sc.parallelize(arr)
-
- //所有的value+1
- val res: RDD[(String,Int)] = rdd.mapValues(value=>value+1)
- res.foreach(println)
运行结果:
- (Spark,2)
- (Hadoop,6)
- (Hive,3)
- (Flink,4)
内连接,(K,V1)和(K,V2)进行内连接生成(K,(V1,V2))。
- //通过并行集合创建RDD
- val arr1 = Array(("Spark",1),("Hadoop",5),("Hive",2),("Flink",3))
- val arr2 = Array(("Spark","fast"),("Hadoop","good"))
- val rdd1: RDD[(String,Int)] = sc.parallelize(arr1)
- val rdd2: RDD[(String,String)] = sc.parallelize(arr2)
-
- //所有的value+1
- // val res: RDD[(String,(Int,Int))] = rdd1.join(rdd2)
- val res: RDD[(String, (Int, String))] = rdd1.join(rdd2)
- res.foreach(println)
运行结果:
- (Spark,(1,fast))
- (Hadoop,(5,good))
我们可以看到,返回的RDD 的元素都是满足连接表rdd2的K的。
这个函数的参数比较多,下面做个介绍:
- //通过并行集合创建RDD
- val arr = Array(("company-1",88),("company-1",96),("company-1",85),("company-2",94),("company-2",86),("company-2",74),("company-3",86),("company-3",88),("company-3",92))
-
- val rdd: RDD[(String, Int)] = sc.parallelize(arr,3)
-
- val res: RDD[(String,Int,Float)] = rdd.combineByKey(
- income=>(income,1),
- (acc:(Int,Int),income)=>(acc._1+income,+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._1/value._2.toFloat)
- })
- //重新分配分区 将3个分区合并为1个
- res.repartition(1).saveAsTextFile("data/kv_rdd/")
运行结果中-part-00000文件内容:
- (company-3,266,88.666664)
- (company-1,269,89.666664)
- (company-2,254,84.666664)
其中,第一列为季度名称。第二列为总收入,第三列为平均收入。
第一个参数的作用是:当我们取出的RDD元素是第一次遇到的key,那么就创建一个组合器函数createCombiner(),负责将我们的键值对(K:季度名称,V:收入额)中的 V:收入额转为 C格式(总收入额,1)的格式,其中的1代表当前已经累加了一个月的收入。
第二个参数是合并值函数 mergeValue(),它的作用是:如果遇到相同的key,比如都是"company-1",那么就对相同key的的value进行mergeValue()中定义的操作。
第三个参数的作用是 :由于我们开启了多个分区,所以最后要对不同分区的数据进行一个对总,这个函数中定义的就是对两个 C格式 的键值对进行的操作。
最后我们进行了一个模式匹配,对于结果返回的(k,v)形式的数据,其中 k 就是指季度名称, v 是一个键值对(总收入额,月份数),我们将它转为 (季度名称,总收入额,平均收入额)。
- 分区1:
- 1-调用createCombiner()函数
- (company-1,88) => (company-1,(88,1))
- 2-调用mergeValue()函数
- (company-1,96) => (company-1,(184,2))
- 分区2:
- 1-调用createCombiner()函数
- (company-1,85) => (company-1,(85,1))
-
- 3-调用mergeCombiners()函数
- (company-1,(184,2)) + (company-1,(85,1)) => (company-1,(269,3))
-
flatMapValues(func)的操作和mapValues(func)相似。它们都是对键值对类型的RDD进行操作,mapValues(func)是对(ke要,value)的value通过函数 func 进行一个处理,而key不变。而flatMapValues(func)则是对value先通过函数 func 进行处理,然后再处理后的值和key组成一系列新的键值对。
输入数据:
- ("k1","hadoop,spark,flink")
- ("k2","hadoop,hive,hbase")
处理
- //通过并行集合创建RDD
- val arr = Array(("k1","hadoop,spark,flink"),("k2","hadoop,hive,hbase"))
- val rdd: RDD[(String, String)] = sc.parallelize(arr)
-
- //flatMapValues(func)
- //val res: Array[(String, String)] = rdd.flatMapValues(value => value.split(",")).collect()
- //mapValues(func)
- val res: Array[(String, Array[String])] =rdd.mapValues(value => value.split(",")).collect()
-
- value.split(",")).collect()
- res.foreach(println)
运行结果:
- (k1,hadoop)
- (k1,spark)
- (k1,flink)
- (k2,hadoop)
- (k2,hive)
- (k2,hbase)
而我们的mapValues(func)执行后的RDD集合内为:
- (k1,Array("hadoop","spark","flink"))
- (k2,Array("hadoop","hive","hbase"))
显然我们的flatMapValues(func)是多进行了一部扁平化的操作,将集合内的元素与key一一组成一系列心得键值对。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。