处理键值对RDD

保存Key/Value对的RDD叫做Pair RDD。

1.创建Pair RDD:

1.1 创建Pair RDD的方式：

很多数据格式在导入RDD时，会直接生成Pair RDD。我们也可以使用map()来将之前讲到的普通RDD转化为Pair RDD。

1.2 Pair RDD转化实例：

下面例子中，把原始RDD，修改成首单词做Key，整行做Value的Pair RDD。

Java中没有tuple类型，所以使用scala的scala.Tuple2类来创建tuple。创建tuple：  new Tuple2(elem1,elem2)  ； 访问tuple的元素：  使用._1()和._2()方法来访问。

而且，在Python和Scala实现中使用基本的map()函数即可，java需要使用函数mapToPair()：

	/**
	 * 将普通的基本RDD转化成一个Pair RDD,业务逻辑： 将每一行的首单词作为Key，整个句子作为Value 返回Key/Value PairRDD。
	 * @param JavaRDD<String>
	 * @return JavaPairRDD<String,String>
	 */
	public JavaPairRDD<String,String> firstWordKeyRdd(JavaRDD<String> input){	
		JavaPairRDD<String,String> pair_rdd = input.mapToPair(
				new PairFunction<String,String,String>(){
					@Override
					public Tuple2<String, String> call(String arg0) throws Exception {
						// TODO Auto-generated method stub
						return new Tuple2<String,String>(arg0.split(" ")[0],arg0);
					}
				}
		);
		return pair_rdd;	
	}

当从内存中的集合创建PairRDD时，Python和Scala需要使用函数SparkContext.parallelize()；而Java使用函数SparkContext.parallelizePairs()。

2.Pair RDD的转化操作:

2.1 Pair RDD常见的转化操作列表：

基础RDD使用的转化操作也可以在Pair RDD中使用。因为Pair RDD中使用tuple，所以需要传递操作tuple的函数给Pair RDD.

下表列出Pair RDD常用的转化操作（事例RDD内容：{(1, 2), (3, 4), (3, 6)}）

下表列举2个RDD之间的转化操作（rdd = {(1, 2), (3, 4), (3, 6)} other = {(3,9)}）：

2.2 Pair RDD筛选操作：

Pair RDD也还是RDD，所以之前介绍的操作（例如filter）也同样适用于PairRDD。下面程序，筛选长度大于20的行：

	/**
	 * PairRDD筛选长度大于20的行。
	 * @param JavaPairRDD<String,String>
	 * @return JavaPairRDD<String,String>
	 */
	public JavaPairRDD<String,String> filterMoreThanTwentyLines
			(JavaPairRDD<String,String> input){	
		JavaPairRDD<String,String> filter_rdd = input.filter(
				new Function<Tuple2<String, String>,Boolean>(){
					@Override
					public Boolean call(Tuple2<String, String> arg0) throws Exception {
						// TODO Auto-generated method stub
						return (arg0._2.length()>20);
					}					
				}
				);
		return filter_rdd;	
	}

2.3 聚合操作：

./spark-shell --jars /spark/alluxio-1.2.0/core/client/target/alluxio-core-client-1.2.0-jar-with-dependencies.jar

Pair RDD提供下面方法：

1. reduceByKey()方法：可以分别归约每个键对应的数据；

2. join()方法：可以把两个RDD中键相同的元素组合在一起，合并为一个RDD。

一、创建Pair RDD：

当需要将一个普通RDD转化成一个Pair RDD时，可以使用map()函数来实现。

程序4-1：   使用第一个单词作为键建出一个Pair RDD：

val text1 = sc.textFile("file:///spark/spark/README.md")

val pair1 = text1.map(x=>( x.split(" ")(0),x))

println(pair1.collect().mkString("  "))

程序4-2：   对Pair RDD的第2个元素筛选：

val text2 = sc.textFile("file:///spark/spark/README.md")

val pair_base2 = text2.map(x=>( x.split(" ")(0),x))

val pair2 = pair_base2 .filter{case(key,value)=>value.length<20}

println(pair2.collect().mkString("  "))

对于二元组数据，有时我们只想访问Pair RDD的值的部分，这时操作二元组很麻烦。可以使用mapValues(func)函数,单操作value，不操作key，功能类似于map{case(x,y):(x,func(y))}

reduceByKey()与reduce()类似，reduceByKey()会为数据集中的每个键进行并行的归约操作，每个归约操作会将键相同的值合并起来。返回一个由各键和对应键归约出来的结果值组成的新的RDD。

foldByKey()使用一个与RDD和合并函数中的数据类型相同的零值作为初始值。

用scala从一个内存中的数据集创建PairRDD时，只需要对这个由二元组组成的集合调用SparkContext.parallelize()方法。

程序4-3：   计算每个键对应的平均值：

val text3 = sc.parallelize(List(("panda",0),("pink",3),("pirate",3),("panda",1),("pink",4)))

val text3_final = text3.mapValues(x=>(x,1)).reduceByKey((x,y)=>(x._1+y._1,x._2+y._2))

println(text3_final.collect().mkString(" "))

调用reduceByKey()和foldByKey()会在每个键计算全局的总结果之前先自动在每台机器上进行本地合并。

其中，mapValues(x=>(x,1))得到的输出结果是："panda",(0,1) ；"pink",(3,1) ；"pirate",(3,1) ；"panda",(1,1) ；"pink",(4,1) ；

下一步，reduceByKey((x,y)=>(x._1+y._1,x._2+y._2))；自动合并同一个key的数据。例如对于panda，(0,1) ，(1,1)   => (0+1 , 1+1) {解释：(0,1)的第一个数据加上(1,1)的第一个数据作为第一个数据，(0,1)的第二个数据加上(1,1)的第二个数据作为第二个数据} 也就是 (1,2)。

程序4-3的输出结果类似于： (pink,(7,2)) (pirate,(3,1)) (panda,(1,2)) ；将每个key的总和得到、并将每个key出现的次数得到。

程序4-4：   实现单词计数：

val text4 = sc.textFile("file:///spark/spark/README.md")

val words = text4.flatMap(x=>x.split(" "))

words.map(x=>(x,1)).reduceByKey((x,y)=>x+y)

combineByKey()是最为常用的基于键进行聚合的函数。combineByKey()可以让用户返回与输入数据的类型不同的返回值。

如果第一次出现一个新的元素（键）（在每一个分区中第一次出现，而不是整个RDD中第一次出现），会使用createCombiner()函数来创建那个键对应的累加器的初始值。

如果是一个在处理当前分区之前已经遇到的键，则使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并。

combineByKey()有多个参数分别对应聚合操作的各个阶段，因而非常适合用来解释聚合操作各个阶段的功能划分。

程序4-5：   使用combineByKey()来实现计算每个key的平均值：

val input = sc.parallelize(List(("panda",0),("pink",3),("pirate",3),("panda",1),("pink",4)))

val result = 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)

).map{case(key,value)=>(key,value._1/value._2.toFloat)}

result.collectAsMap().map(println(_))

其中combineByKey()接收3个函数：

第一个函数是createCombiner()，也就是在某一个分区第一次碰到一个新的key，比如panda是第一次出现，则调用createCombiner函数。

第二个函数是mergeValue()，在同一个分区中，如果出现了之前出现过的一个key，例如是panda，这时候调用这个函数，

第三个函数是mergeCombiners()，