spark-java+scala 乐字节教育案例

清疚 2023-01-14 13:58 41阅读 0赞

# 1.统计PV&UV #

pv某个网页24小时被点击的次数  
uv每个网页被不同用户点击的次数  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]

数据以制表符隔开"\\t"

146.1.30.98	河南	2017-10-10	1512012307078	5263761960810313758	www.jd.com	Regist
    146.1.30.98	河南	2017-10-10	1512012307079	5263761960810313758	www.taobao.com	Login
    146.1.30.98	河南	2017-10-10	1512012307080	5263761960810313758	www.baidu.com	View
    146.1.30.98	河南	2017-10-10	1512012307080	5263761960810313758	www.jd.com	Comment
    146.1.30.98	河南	2017-10-10	1512012307080	5263761960810313758	www.mi.com	View
    98.188.31.30	上海	2017-10-10	1512012307081	2605157034872986714	www.mi.com	View
    192.149.67.17	天津	2017-10-10	1512012307081	539962031835013115	www.baidu.com	Comment
    73.91.219.62	青海	2017-10-10	1512012307081	6423010865035708629	www.gome.com.cn	Login
    73.91.219.62	青海	2017-10-10	1512012307082	6423010865035708629	www.dangdang.com	Click
    73.91.219.62	青海	2017-10-10	1512012307082	6423010865035708629	www.gome.com.cn	Buy
    197.40.115.113	上海	2017-10-10	1512012307083	5517762668894634404	www.gome.com.cn	Buy
    197.40.115.113	上海	2017-10-10	1512012307083	5517762668894634404	www.gome.com.cn	Login
    197.40.115.113	上海	2017-10-10	1512012307083	5517762668894634404	www.gome.com.cn	Regist
    197.40.115.113	上海	2017-10-10	1512012307083	5517762668894634404	www.taobao.com	Buy

package com.shsxt.java.core.demo;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.Function2;
    import org.apache.spark.api.java.function.PairFunction;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    public class lcs_PV { 
        public static void main(String[] args) { 
    
            SparkConf sparkConf = new SparkConf();
            sparkConf.setMaster("local").setAppName("pv");
            JavaSparkContext context = new JavaSparkContext(sparkConf);
    
            JavaRDD<String> lineRDD = context.textFile("./data/pvuvdata");
    
    
            /** * 每个页面 PV */
    
            lineRDD.mapToPair(new PairFunction<String, String, Integer>() { 
                @Override
                public Tuple2<String, Integer> call(String line) throws Exception { 
                    String[] str = line.split("\t");
    
    
                    return new Tuple2<>(str[5],1);
                }
            }).reduceByKey(new Function2<Integer, Integer, Integer>() { 
                @Override
                public Integer call(Integer v1, Integer v2) throws Exception { 
                        return v1 + v2;
                }
            }).foreach(new VoidFunction<Tuple2<String, Integer>>() { 
                @Override
                public void call(Tuple2<String, Integer> tuple2) throws Exception { 
                    System.out.println(tuple2);
                }
            });
            
        }
    }

package com.shsxt.java.core.demo;
    
    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.PairFunction;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    import java.util.HashSet;
    import java.util.Iterator;
    import java.util.Map;
    
    public class Lcs_UV { 
        private  int sum = 10;
    
    
        public static void main(String[] args) { 
            SparkConf sparkConf = new SparkConf();
            sparkConf.setMaster("local").setAppName("pv");
            JavaSparkContext context = new JavaSparkContext(sparkConf);
    
            JavaRDD<String> lineRDD = context.textFile("./data/pvuvdata");
    //方法1
            JavaPairRDD<String, Iterable<String>> rdd1 = lineRDD.mapToPair(new PairFunction<String, String, String>() { 
                @Override
                public Tuple2<String, String> call(String s) throws Exception { 
    
    
                    String url = s.split("\t")[5];
                    String ip = s.split("\t")[0];
                    return new Tuple2<>(url, ip);
                }
            }).groupByKey();
    
            rdd1.foreach(new VoidFunction<Tuple2<String, Iterable<String>>>() { 
                @Override
                public void call(Tuple2<String, Iterable<String>> tuple2) throws Exception { 
    
    
                    HashSet<Object> set = new HashSet<>();
    
                    Iterator<String> iterator = tuple2._2.iterator();
                    //用set集合去重
                    while(iterator.hasNext()){ 
                        set.add(iterator.next());
                    }
    
                    System.out.println(tuple2._1  + " : " + set.size());
                }
            });
    
    //方法2
            Map<String, Long> map = lineRDD.mapToPair(new PairFunction<String, String, String>() { 
                @Override
                public Tuple2<String, String> call(String s) throws Exception { 
    
                    String url = s.split("\t")[5];
                    String ip = s.split("\t")[0];
                    return new Tuple2<>(url, ip);
                }
            }).distinct().countByKey();
    
            for (String key : map.keySet()) { 
                System.out.println("key : " + key + " value :" + map.get(key));
            }
    
    
        }
    }

# 2.二次排序 #

先对第一个数据排序在对第二个数据排序  
数据以空格隔开

1 3
    1 4
    2 3
    3 1
    6 7
    2 9
    5 6
    1 9
    5 6
    4 4
    2 3
    1 9
    4 4
    3 6

自定义一个可比较的类为key，使用这个类对象来sortKey  
自定义key实现序列化和comparable接口

package com.shsxt.java.core.demo;
    
    import java.io.Serializable;
    
    public class SecondSortKey  implements Serializable, Comparable<SecondSortKey> { 
    
        private static final long serialVersionUID = 1L;
        //根据数据格式定时成员属性
        private int first;
        private int second;
        public int getFirst() { 
            return first;
        }
        public void setFirst(int first) { 
            this.first = first;
        }
        public int getSecond() { 
            return second;
        }
        public void setSecond(int second) { 
            this.second = second;
        }
    
        public SecondSortKey(int first, int second) { 
            super();
            this.first = first;
            this.second = second;
        }
    
    //重写compareTo方法
        @Override
        public int compareTo(SecondSortKey o1) { 
    //如果第一个数相同，比较第二个
            if (getFirst() - o1.getFirst() == 0) { 
                // 5 6
                // this < o1
                // 6 5
                // this > o1
                return o1.getSecond() - getSecond();
    //比较第3个
                /*if(getSecond() - o1.getSecond() == 0) { return getThree() - o1.getThree(); }else { return getSecond() - o1.getSecond(); }*/
    
            } else { 
                //如果第一个数不相同
                return  o1.getFirst() - getFirst();
        }
    
        }
    }

package com.shsxt.java.core.demo;
    
    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.PairFunction;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    //
    public class SecondKeyTest { 
    
        public static void main(String[] args) { 
            SparkConf sparkConf = new SparkConf();
    
            sparkConf.setMaster("local")
                    .setAppName("SecondarySortTest");
            final JavaSparkContext sc = new JavaSparkContext(sparkConf);
    
            JavaRDD<String> secondRDD = sc.textFile("data/secondSort.txt");
    
            JavaPairRDD<SecondSortKey, String> secRDD = secondRDD.mapToPair(new PairFunction<String, SecondSortKey, String>() { 
                @Override
                public Tuple2<SecondSortKey, String> call(String line) throws Exception { 
                    String[] fields = line.split(" ");
                    SecondSortKey secondSortKey = new SecondSortKey(Integer.parseInt(fields[0]), Integer.parseInt(fields[1]));
                    System.out.println("SecondSortKey:"+secondSortKey);
                    //使用自定义的可比较的类对象为key,使用这个类对象来sortKey
                    return new Tuple2<>(secondSortKey, line);
                }
            });
    
            secRDD.sortByKey().foreach(new VoidFunction<Tuple2<SecondSortKey, String>>() { 
                @Override
                public void call(Tuple2<SecondSortKey, String> tuple2) throws Exception { 
                    System.out.println(tuple2._2);
                }
            });
        }
    }

# 3. 分组取 topN 和 topN #

分组后取每个组里面的前几个

数据

a 86
    a 58
    b 78
    c 47
    a 17
    b 51
    b 98
    c 38
    d 23
    d 12
    m 48
    a 12
    a 84
    b 23
    b 12
    c 89
    c 73

package com.shsxt.java.core.demo;
    
    
    import org.apache.commons.collections.IteratorUtils;
    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.PairFunction;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    import java.util.ArrayList;
    import java.util.Collections;
    import java.util.Iterator;
    import java.util.List;
    
    
    public class TopNTest { 
    
        public static void main(String[] args) { 
    
            SparkConf conf;
            conf = new SparkConf()
                    .setMaster("local[5]")
                    .setAppName("TopOps");
    
    
            JavaSparkContext sc = new JavaSparkContext(conf);
            //hdfs://shsxt/wc.txt
            JavaRDD<String> linesRDD = sc.textFile("data/scores.txt",5);
    
            final List n = new ArrayList();
    // linesRDD.count();
    
            JavaPairRDD<String, Integer> pairRDD = linesRDD.mapToPair(
                    new PairFunction<String, String, Integer>() { 
    
                        /** * */
                        private static final long serialVersionUID = 1L;
    
                        @Override
                        public Tuple2<String, Integer> call(String str) throws Exception { 
    
                            String[] splited = str.split(" ");
                            String clazzName = splited[0];
                            Integer score = Integer.valueOf(splited[1]);
                            return new Tuple2<String, Integer>(clazzName, score);
                        }
                    });
    
    
    
            JavaPairRDD<String, Iterable<Integer>> groupByKeyRDD = pairRDD.groupByKey();
    
          //方法2 使用collections集合提供的List排序方法，默认升序
          groupByKeyRDD.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() { 
            @Override
            public void call(Tuple2<String, Iterable<Integer>> tuple2) throws Exception { 
              String key  = tuple2._1;
              Iterator<Integer> iterator = tuple2._2.iterator();
              List list = IteratorUtils.toList(iterator);
              Collections.sort(list);
              //升序取出前3个，降序取出后3个
              for(int i=0;i<Math.min(3,list.size());i++){ 
                // list.size = 3 list.get(2)
                System.out.println(key + " " +  list.get(list.size()-i-1));
              }
            }
          });
          
          //方法1 iterator比较大小
            groupByKeyRDD.foreach(
                    new VoidFunction<Tuple2<String, Iterable<Integer>>>() { 
    
                        private static final long serialVersionUID = 1L;
    
                        @Override
                        public void call(Tuple2<String, Iterable<Integer>> tuple) throws Exception { 
    
                            String clazzName = tuple._1;
                            Iterator<Integer> iterator = tuple._2.iterator();
                            System.out.println(tuple);
    
                            Integer[] top3 = new Integer[3];
    
                            while (iterator.hasNext()) { 
    
                                Integer score = iterator.next();
    
                                for (int i = 0; i < top3.length; i++) { 
                                    if (top3[i] == null) { 
                                        top3[i] = score;
                                        break;
                                    } else if (score > top3[i]) { 
                                        for (int j = 2; j > i; j--) { 
                                            top3[j] = top3[j - 1];
                                        }
                                        top3[i] = score;
                                        break;
                                    }
                                }
                            }
                            System.out.println("class Name:" + clazzName);
                            for (Integer sscore : top3) { 
                                System.out.println(sscore);
                            }
    
                        }
                    });
    
    
        }
    }

p23

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]: /images/20221022/76b96202ebde4c3b85189f47b91abbb9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]: /images/20221022/520e690f99204d7e95595c30f80873ef.png