Spark实现分组取 top N 示例 —— Java版

电玩女神 2022-05-18 03:48 279阅读 0赞

主要步骤：

1.  加载集合生成RDD（textFile）
2.  对RDD进行转换，将要排序的属性进行分离，生成新的RDD（mapToPair）
3.  对键进行分组，并在分组内基于第二步分离出的属性进行排序，并取排序结果的 top N （groupByKey，mapToPair）

package rddDemo.examples;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaPairRDD;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.api.java.function.PairFunction;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    import java.util.*;
    
    /**
     * Created by asus on 2018/7/29.
     * 数据样本格式如下：
         anhui hefei 20
         jiangsu nanjing 90
         shandong jinan 100
     * 分组取 top N
     */
    public class TopNWithGroupDemoJava {
        public static void main(String[] args ) {
            SparkConf conf = new SparkConf() ;
            conf.setAppName("TopNWithGroupDemoJava") ;
            conf.setMaster("local[2]") ;
    
            System.setProperty("hadoop.home.dir" , "E:\\hadoop-2.6.0") ;
    
            final JavaSparkContext sc = new JavaSparkContext(conf) ;
            String filepath = "hdfs://192.168.204.130:9000/log_file/sale_data.txt" ;
    
            // 加载 HDFS 文件构造RDD，并使用 filter 算子去掉空行
            JavaRDD<String> saleRDD = sc.textFile(filepath , 2).filter(new Function<String, Boolean>() {
                @Override
                public Boolean call(String v1) throws Exception {
                    return v1.trim().length() > 0 ;
                }
            }) ;
            // 对RDD进行转换，将要排序的属性作为键
            JavaPairRDD<String , Tuple2<Integer , String>> saleWithProvinceRDD = saleRDD.mapToPair(new PairFunction<String, String, Tuple2<Integer, String>>() {
                @Override
                public Tuple2<String, Tuple2<Integer, String>> call(String s) throws Exception {
                    if(s.trim().length() != 0) {
                        String[] infos = s.split("\\s+") ;
                        String province = infos[0] ;
                        int saleCount = Integer.parseInt(infos[2]) ;
                        return new Tuple2<>(province , new Tuple2<>(saleCount, s));
                    }else {
                        return null ;
                    }
                }
            }) ;
    //        saleWithProvinceRDD.foreach(new VoidFunction<Tuple2<String, Tuple2<Integer, String>>>() {
    //            @Override
    //            public void call(Tuple2<String, Tuple2<Integer, String>> t) throws Exception {
    //                if(t != null) {
    //                    System.out.println("province : " + t._1 + " saleCount : " + t._2._1 + " info : " + t._2._2);
    //                }
    //            }
    //        });
    
            // 进行分组
            JavaPairRDD<String , Iterable<Tuple2<Integer , String>>> saleGroupByProvinceRDD = saleWithProvinceRDD.groupByKey() ;
            // 在分组内进行排序，取分组内的 top N
            JavaPairRDD<String , Iterable<Tuple2<Integer , String>>> saleGroupByProvinceTopNRDD = saleGroupByProvinceRDD.mapToPair(new PairFunction<Tuple2<String,Iterable<Tuple2<Integer,String>>>, String, Iterable<Tuple2<Integer,String>>>() {
                private static final long serialVersionID = 1L ;
                @Override
                public Tuple2<String, Iterable<Tuple2<Integer,String>>> call(Tuple2<String, Iterable<Tuple2<Integer, String>>> t) throws Exception {
                    // 取出键
                    String province = t._1 ;
                    // 记录每个键对应的分组内容
                    List<Tuple2<Integer , String>> saleInfoList = new ArrayList<>() ;
                    Iterable<Tuple2<Integer , String>> saleInfo = t._2 ;
                    for(Tuple2<Integer , String> temp : saleInfo) {
                        saleInfoList.add(new Tuple2<>(temp._1 , temp._2)) ;
                    }
    
                    // 对分组内容进行自定义排序，数据量大时 Collections.sort 会出现 OOM 内存溢出，建议用插入排序或者归并排序
                    Collections.sort(saleInfoList, new Comparator<Tuple2<Integer, String>>() {
                        @Override
                        public int compare(Tuple2<Integer, String> t1, Tuple2<Integer, String> t2) {
                            if(t1 != null && t2 != null) {
                                return -(t1._1 - t2._1);
                            }
                            return 0 ;
                        }
                    });
                    saleInfoList = saleInfoList.subList(0, 3) ;
    
                    /**
                     * 以下代码不可行，call 方法在 executor 中进行调用，executor中不能再创建RDD
                     */
    //                JavaPairRDD<Integer , String> saleInfoRDD = sc.parallelizePairs(saleInfoList) ;
    //                List<Tuple2<Integer , String>> res = saleInfoRDD.sortByKey(false).take(3) ;
    
                    return new Tuple2<String , Iterable<Tuple2<Integer,String>>>(province , saleInfoList) ;
                }
            }) ;
    
            saleGroupByProvinceTopNRDD.foreach(new VoidFunction<Tuple2<String, Iterable<Tuple2<Integer, String>>>>() {
                @Override
                public void call(Tuple2<String, Iterable<Tuple2<Integer, String>>> t1) throws Exception {
                    if(t1 != null) {
                        for(Tuple2<Integer , String> s : t1._2) {
                            System.out.println("province : " + t1._1 + " saleCount : " + s._1 + " info : " + s._2);
                        }
                    }
                }
            });
    
            sc.stop();
        }
    }