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 {
                //過濾合法和非靜態(tài)的資源的日志
                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("總記錄數(shù):"+lines.count());
        // 統(tǒng)計ip
            ipsHandle(lines);
        System.out.println("--------------------------------------------------");
        // 統(tǒng)計分時頁面訪問量
            timeHandle(lines);
        System.out.println("--------------------------------------------------");
        // 統(tǒng)計分時ip訪問量
            timeHandle2(lines);
        System.out.println("--------------------------------------------------");
        // 統(tǒng)計狀態(tài) 400，300 200 各有多少,計算失敗率
            stateHandle(lines);
        System.out.println("--------------------------------------------------");
        // 統(tǒng)計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數(shù):"+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);

            }
        });

    }

}