/**
 * Analyze log with Spark.
 */
public class AccessLogAnalyzer {

    private static final SimpleDateFormat logDateFormat = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z", Locale.US);
    private static final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMdd");

    private static String fetchDate(String gmsTime) {
        String date;
        try {
            date = simpleDateFormat.format(logDateFormat.parse(gmsTime));
        } catch (ParseException e) {
            date = null;
        }
        return date;
    }

    private static final Pattern LOGPATTERN = Pattern.compile(
            "(\\S+)\\s+-\\s+\\[(.*?)\\]\\s+\"(.*?)\"\\s+(\\d{3})\\s+(\\S+)\\s+"
            + "\"(.*?)\"\\s+\"(.*?)\"\\s+(.*?)\\s+\"(.*?)\"\\s+(\\S+)\\s+(\\S+)\\s+(\\S+)");

    private static Tuple2<String, String> extractKey(String line) {
        Matcher m = LOGPATTERN.matcher(line);
        if (m.find()) {
            String ipAddr = m.group(1);
            String date = fetchDate(m.group(2));
            return new Tuple2<>(date, ipAddr);
        }
        return new Tuple2<>(null, null);
    }

    public static void main(String[] args) {
        if (args.length != 3) {
            System.err.println("usage: spark-submit com.baidu.cloud.bmr.spark.AccessLogAnalyzer <input> <pv> <uv>");
            System.exit(1);
        }

        SparkConf conf = new SparkConf().setAppName("AccessLogAnalyzer");
        JavaSparkContext sc = new JavaSparkContext(conf);

        // Parses the log to log records and caches the result.
        JavaPairRDD<String, String> distFile = sc.textFile(args[0]).mapToPair(
                new PairFunction<String, String, String>() {
                    @Override
                    public Tuple2<String, String> call(String s) {
                        return extractKey(s);
                    }
                });
        distFile.cache();

        // Changes the log info to (date, 1) format, and caculates the page view.
        JavaPairRDD<String, Integer> pv = distFile.mapToPair(
                new PairFunction<Tuple2<String, String>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<String, String> tuple) {
                        return new Tuple2<>(tuple._1(), 1);
                    }
                }).reduceByKey(new Function2<Integer, Integer, Integer>() {
                    @Override
                    public Integer call(Integer i1, Integer i2) {
                        return i1 + i2;
                    }
                });
        // Coalesces to 1 partition and saves as file. Notice that this is for demo purpose only.
        pv.coalesce(1, true).saveAsTextFile(args[1]);

        // Changes the log info to (date, remoteAddr) and caculates the unique visitors.
        JavaPairRDD<String, Integer> uv = distFile.groupByKey().mapToPair(
                new PairFunction<Tuple2<String, Iterable<String>>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<String, Iterable<String>> tuple) {
                        int size = new HashSet((Collection<?>) tuple._2()).size();
                        return new Tuple2<>(tuple._1(), size);
                    }
                });
        // Coalesces to 1 partition and saves as file. Notice that this is for demo purpose only.
        uv.coalesce(1, true).saveAsTextFile(args[2]);
    }
}

------PV------
20151003    139
20151005    372
20151006    114
20151004    375
------UV------
20151003    111
20151005    212
20151006    97
20151004    247

-----PV-----
(20151003,139)
(20151005,372)
(20151006,114)
(20151004,375)
-----UV-----
(20151003,111)
(20151005,212)
(20151006,97)
(20151004,247)

------PV------
+--------+---+
|    date| pv|
+--------+---+
|20151003|139|
|20151004|375|
|20151005|372|
|20151006|114|
+--------+---+

------UV------
+--------+---+
|    date| uv|
+--------+---+
|20151003|111|
|20151004|247|
|20151005|212|
|20151006| 97|
+--------+---+