（九）groupByKey,reduceByKey,sortByKey算子-Java&Python版Spark

groupByKey,reduceByKey,sortByKey算子

视频教程：

1、优酷

2、 YouTube

1、groupByKey

groupByKey是对每个key进行合并操作，但只生成一个sequence，groupByKey本身不能自定义操作函数。

java：

 package com.bean.spark.trans;

 import java.util.Arrays;

 import java.util.List;

 import org.apache.spark.SparkConf;

 import org.apache.spark.api.java.JavaPairRDD;

 import org.apache.spark.api.java.JavaSparkContext;

 import scala.Tuple2;

 public class TraGroupByKey {

     public static void main(String[] args) {

         SparkConf conf = new SparkConf();

         conf.setMaster("local");

         conf.setAppName("union");

         System.setProperty("hadoop.home.dir", "D:/tools/spark-2.0.0-bin-hadoop2.6");

         JavaSparkContext sc = new JavaSparkContext(conf);

         List<Tuple2<String, Integer>> list = Arrays.asList(new Tuple2<String, Integer>("cl1", 90),

                 new Tuple2<String, Integer>("cl2", 91),new Tuple2<String, Integer>("cl3", 97),

                 new Tuple2<String, Integer>("cl1", 96),new Tuple2<String, Integer>("cl1", 89),

                 new Tuple2<String, Integer>("cl3", 90),new Tuple2<String, Integer>("cl2", 60));

         JavaPairRDD<String, Integer> listRDD = sc.parallelizePairs(list);

         JavaPairRDD<String, Iterable<Integer>> results  = listRDD.groupByKey();

         System.out.println(results.collect());

         sc.close();

     }

 }

python：

 # -*- coding:utf-8 -*-

 from pyspark import SparkConf

 from pyspark import SparkContext

 import os

 if __name__ == '__main__':

     os.environ["SPARK_HOME"] = "D:/tools/spark-2.0.0-bin-hadoop2.6"

     conf = SparkConf().setMaster('local').setAppName('group')

     sc = SparkContext(conf=conf)

     data = [('tom',90),('jerry',97),('luck',92),('tom',78),('luck',64),('jerry',50)]

     rdd = sc.parallelize(data)

     print rdd.groupByKey().map(lambda x: (x[0],list(x[1]))).collect()

注意：当采用groupByKey时，由于它不接收函数，spark只能先将所有的键值对都移动，这样的后果是集群节点之间的开销很大，导致传输延时。

整个过程如下：

因此，在对大数据进行复杂计算时，reduceByKey优于groupByKey。

另外，如果仅仅是group处理，那么以下函数应该优先于 groupByKey ：

（1）、combineByKey 组合数据，但是组合之后的数据类型与输入时值的类型不一样。

（2）、foldByKey合并每一个 key 的所有值，在级联函数和“零值”中使用。

2、reduceByKey

对数据集key相同的值，都被使用指定的reduce函数聚合到一起。

java：

 package com.bean.spark.trans;

 import java.util.Arrays;

 import java.util.List;

 import org.apache.spark.SparkConf;

 import org.apache.spark.api.java.JavaPairRDD;

 import org.apache.spark.api.java.JavaSparkContext;

 import org.apache.spark.api.java.function.Function2;

 import scala.Tuple2;

 public class TraReduceByKey {

     public static void main(String[] args) {

         SparkConf conf = new SparkConf();

         conf.setMaster("local");

         conf.setAppName("reduce");

         System.setProperty("hadoop.home.dir", "D:/tools/spark-2.0.0-bin-hadoop2.6");

         JavaSparkContext sc = new JavaSparkContext(conf);

         List<Tuple2<String, Integer>> list = Arrays.asList(new Tuple2<String, Integer>("cl1", 90),

                 new Tuple2<String, Integer>("cl2", 91),new Tuple2<String, Integer>("cl3", 97),

                 new Tuple2<String, Integer>("cl1", 96),new Tuple2<String, Integer>("cl1", 89),

                 new Tuple2<String, Integer>("cl3", 90),new Tuple2<String, Integer>("cl2", 60));

         JavaPairRDD<String, Integer> listRDD = sc.parallelizePairs(list);

         JavaPairRDD<String, Integer> results = listRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {

             @Override

             public Integer call(Integer s1, Integer s2) throws Exception {

                 // TODO Auto-generated method stub

                 return s1 + s2;

             }

         });

         System.out.println(results.collect());

 sc.close();

 }

 }

python：

 # -*- coding:utf-8 -*-

 from pyspark import SparkConf

 from pyspark import SparkContext

 import os

 from operator import add

 if __name__ == '__main__':

     os.environ["SPARK_HOME"] = "D:/tools/spark-2.0.0-bin-hadoop2.6"

     conf = SparkConf().setMaster('local').setAppName('reduce')

     sc = SparkContext(conf=conf)

     data = [('tom',90),('jerry',97),('luck',92),('tom',78),('luck',64),('jerry',50)]

     rdd = sc.parallelize(data)

 print rdd.reduceByKey(add).collect()

 sc.close()

当采用reduceByKey时，Spark可以在每个分区移动数据之前将待输出数据与一个共用的key结合。 注意在数据对被搬移前同一机器上同样的key是怎样被组合的。

3、sortByKey

通过key进行排序。

java:

 package com.bean.spark.trans;

 import java.util.Arrays;

 import java.util.List;

 import org.apache.spark.SparkConf;

 import org.apache.spark.api.java.JavaPairRDD;

 import org.apache.spark.api.java.JavaSparkContext;

 import scala.Tuple2;

 public class TraSortByKey {

     public static void main(String[] args) {

         SparkConf conf = new SparkConf();

         conf.setMaster("local");

         conf.setAppName("sort");

         System.setProperty("hadoop.home.dir", "D:/tools/spark-2.0.0-bin-hadoop2.6");

         JavaSparkContext sc = new JavaSparkContext(conf);

         List<Tuple2<Integer, String>> list = Arrays.asList(new Tuple2<Integer,String>(3,"Tom"),

                 new Tuple2<Integer,String>(2,"Jerry"),new Tuple2<Integer,String>(5,"Luck")

                 ,new Tuple2<Integer,String>(1,"Spark"),new Tuple2<Integer,String>(4,"Storm"));

         JavaPairRDD<Integer,String> rdd = sc.parallelizePairs(list);

         JavaPairRDD<Integer, String> results = rdd.sortByKey(false);

         System.out.println(results.collect());

       sc.close()

     }

 }    

python:

 #-*- coding:utf-8 -*-

 if __name__ == '__main__':

     os.environ["SPARK_HOME"] = "D:/tools/spark-2.0.0-bin-hadoop2.6"

     conf = SparkConf().setMaster('local').setAppName('reduce')

     sc = SparkContext(conf=conf)

     data = [(5,90),(1,92),(3,50)]

     rdd = sc.parallelize(data)

 print rdd.sortByKey(False).collect()

 sc.close()