Onelong

分享知识,与你一起进步......
RSS icon Home icon
  • spark java demo

    post by onelong / 2016-11-18 18:57 Friday [工作]

    import java.text.SimpleDateFormat;
    import java.util.*;

    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.*;

    import scala.Tuple2;

    public class SparkDemo {
     
        static final String USER = "huanglong";
     
        public static void main(String[] args) throws Exception {
            
           
            SparkConf conf =new SparkConf().setAppName("Spark Log").setMaster("local[4]");
            JavaSparkContext sc = new JavaSparkContext(conf); //其底层就是scala的sparkcontext


            //JavaSparkContext sc = new JavaSparkContext("local[4]", "Spark Log", "/user/huanglong/spark", new String[0], envs);
     
           
            String file="file:///Users/huanglong/access_05_30.log";
            JavaRDD<String> data = sc.textFile(file, 4).cache();

            //日志格式
            //27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/faq.gif HTTP/1.1" 200 1127;

            JavaRDD<String> filter =  data.filter(new org.apache.spark.api.java.function.Function<String, Boolean>() {
                @Override
                public Boolean call(String s) throws Exception {
                    //过滤合法和非静态的资源的日志
                    return (s.contains("POST")||s.contains("GET"))&& !s.contains("/static/");
                }
            });

            final  SimpleDateFormat tf = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z",
                    Locale.ENGLISH);
            //只有一天的记录,不需要年月日了,时间精度到分钟
            final SimpleDateFormat tf1 = new SimpleDateFormat("HH:mm");

            JavaRDD<Map<String,String>> lines = filter.flatMap(new FlatMapFunction<String,Map<String,String>>(){

                @Override
                public Iterator<Map<String,String>> call(String s) throws Exception {
                    List<Map<String,String>> list = new ArrayList<Map<String,String>>(1);
                    Map<String,String> map = new HashMap<String, String>();
                    list.add(map);
                    String ip = s.substring(0,s.indexOf(" - - "));
                    map.put("ip",ip);

                    String t = s.substring(s.indexOf("[")+1,s.indexOf("]"));
                    Date date =  tf.parse(t);
                    map.put("time",tf1.format(date));

                    String http = s.substring(s.indexOf("\"")+1,s.indexOf("\" "));
                    String[] h= http.split(" ");

                    if(h.length>1){

                        if(h[1]!=null&&h[1].indexOf("?")!=-1)
                        {
                            map.put("url",h[1].substring(0,h[1].indexOf("?")));
                        }else {
                            map.put("url",h[1]);
                        }
                    }

                    if(h.length>2)
                    {
                        map.put("version",h[2]);
                    }

                    if(map.get("version")!=null&&map.get("version").contains("1.1"))
                    {
                        String state = s.substring(s.indexOf("1.1\" ")+5,s.length());
                        String[] strings = state.split(" ");
                        if(strings.length>0) {
                            map.put("state", strings[0]);
                        }
                    }else if(map.get("version")!=null&&map.get("version").contains("1.0"))
                    {
                        String state = s.substring(s.indexOf("1.0\" ")+5,s.length());
                        String[] strings = state.split(" ");
                        if(strings.length>0) {
                            map.put("state", strings[0]);
                        }
                    }else {
                        //System.out.println(s);
                    }

                    return list.iterator();
                }
            });

            System.out.println("总记录数:"+lines.count());
            // 统计ip
                ipsHandle(lines);
            System.out.println("--------------------------------------------------");
            // 统计分时页面访问量
                timeHandle(lines);
            System.out.println("--------------------------------------------------");
            // 统计分时ip访问量
                timeHandle2(lines);
            System.out.println("--------------------------------------------------");
            // 统计状态 400,300 200 各有多少,计算失败率
                stateHandle(lines);
            System.out.println("--------------------------------------------------");
            // 统计http版本 1.1、1.0
                versionHandle(lines);
            // 页面热度
            System.out.println("--------------------------------------------------");
                pageHandle(lines);
     
           sc.close();

        }


        private static void pageHandle(JavaRDD<Map<String, String>> lines) {

            JavaRDD<Map<String, String>> filterData =  lines.filter(new Function<Map<String, String>, Boolean>() {
                @Override
                public Boolean call(Map<String, String> map) throws Exception {
                    String url = map.get("url");
                    return "200".equals(map.get("state"))&&url.endsWith(".php");
                }
            });


            JavaPairRDD<String,Integer> versionPairs = filterData.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
                @Override
                public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
                    return  new Tuple2<String, Integer>(stringStringMap.get("url"),1);
                }
            });

            JavaPairRDD<String,Integer> versionCount = versionPairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
                @Override
                public Integer call(Integer v1, Integer v2) throws Exception {
                    return v1+v2;
                }
            });


            versionCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
                @Override
                public void call(Tuple2<String, Integer> pairs) throws Exception {
                    System.out.println("URL->PV:"+pairs._1+" : " +pairs._2);
                }
            });
        }


        private static void versionHandle(JavaRDD<Map<String, String>> lines) {
            JavaPairRDD<String,Integer> versionPairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
                @Override
                public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
                    return  new Tuple2<String, Integer>(stringStringMap.get("version"),1);
                }
            });

            JavaPairRDD<String,Integer> versionCount = versionPairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
                @Override
                public Integer call(Integer v1, Integer v2) throws Exception {
                    return v1+v2;
                }
            });


            versionCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
                @Override
                public void call(Tuple2<String, Integer> pairs) throws Exception {
                    System.out.println("VERSION->PV:"+pairs._1+" : " +pairs._2);
                }
            });
        }

        private static void stateHandle(JavaRDD<Map<String, String>> lines) {
            JavaPairRDD<String,Integer> statePairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
                @Override
                public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
                    return  new Tuple2<String, Integer>(stringStringMap.get("state"),1);
                }
            });

            JavaPairRDD<String,Integer> stateCount = statePairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
                @Override
                public Integer call(Integer v1, Integer v2) throws Exception {
                    return v1+v2;
                }
            });

            stateCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
                @Override
                public void call(Tuple2<String, Integer> pairs) throws Exception {
                    System.out.println("STATE->PV:"+pairs._1+" : " +pairs._2);

                }
            });
        }

        private static void ipsHandle(JavaRDD<Map<String, String>> lines) {
            JavaPairRDD<String,Integer> ipPairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
                @Override
                public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
                    return  new Tuple2<String, Integer>(stringStringMap.get("ip"),1);
                }
            });

            JavaPairRDD<String,Integer> ipCount = ipPairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
                @Override
                public Integer call(Integer v1, Integer v2) throws Exception {
                    return v1+v2;
                }
            });

            System.out.println("总ip数:"+ipCount.count());


            ipCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
                @Override
                public void call(Tuple2<String, Integer> pairs) throws Exception {
                    System.out.println("IP->PV:"+pairs._1+" : " +pairs._2);
                }
            });
        }

        private static void timeHandle2(JavaRDD<Map<String, String>> lines) {
            JavaPairRDD<String,Map<String,Boolean>> timePairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Map<String,Boolean>>() {
                @Override
                public Tuple2<String,Map<String,Boolean>> call(Map<String, String> stringStringMap) throws Exception {
                    Map<String,Boolean> m = new HashMap<String, Boolean>();
                    m.put(stringStringMap.get("ip"),Boolean.TRUE);
                    return new Tuple2<String, Map<String,Boolean>>(stringStringMap.get("time"),m);
                }
            });

            JavaPairRDD<String,Map<String,Boolean>> timeCount = timePairs.reduceByKey(new Function2<Map<String,Boolean>, Map<String,Boolean>,Map<String,Boolean>>() {
                @Override
                public Map<String,Boolean> call(Map<String,Boolean> v1, Map<String,Boolean> v2) throws Exception {
                    HashMap<String, Boolean> m = new HashMap<String, Boolean>();
                    m.putAll(v1);
                    m.putAll(v2);
                    return m;
                }
            });


            timeCount.foreach(new VoidFunction<Tuple2<String,Map<String,Boolean>>>() {
                @Override
                public void call(Tuple2<String, Map<String,Boolean>> pairs) throws Exception {

                    System.out.println("TIME->IP:"+pairs._1+" : " +pairs._2.size());

                }
            });

        }

        private static void timeHandle(JavaRDD<Map<String, String>> lines) {
            JavaPairRDD<String,Integer> timePairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
                @Override
                public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
                    return  new Tuple2<String, Integer>(stringStringMap.get("time"),1);
                }
            });

            JavaPairRDD<String,Integer> timeCount = timePairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
                @Override
                public Integer call(Integer v1, Integer v2) throws Exception {
                    return v1+v2;
                }
            });

            timeCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
                @Override
                public void call(Tuple2<String, Integer> pairs) throws Exception {
                    System.out.println("TIME->PV:"+pairs._1+" : " +pairs._2);

                }
            });

        }

    }

    更多请关注 https://github.com/OnelongX/demos

    引用地址:
     

    我要评论