// This program aggregates by IP hits on apache access logs. 
// It can get the top ten of IP's
//

package com.amazonaws.vivanih.hadoop.cascading;

import cascading.flow.Flow;
import cascading.flow.FlowDef;
import cascading.flow.hadoop.HadoopFlowConnector;
import cascading.operation.aggregator.Count;
import cascading.operation.filter.Sample;
import cascading.operation.filter.Limit;
import cascading.operation.regex.RegexParser;
import cascading.operation.text.DateParser;
import cascading.pipe.*;
import cascading.property.AppProps;
import cascading.scheme.hadoop.TextDelimited;
import cascading.scheme.hadoop.TextLine;
import cascading.tap.SinkMode;
import cascading.tap.Tap;
import cascading.tap.hadoop.Hfs;
import cascading.tuple.Fields;
import java.util.Properties;
 
public class Main {
    public static void main(String[] args) {
 
        String inputPath 		= args[0]; //input will use default filesystem. HDFS or S3.
        String outputPath 	= args[1]; 
       // String sortedTopTen = args[ 1 ] + "/top10/";
        
        
        // sources and sinks
        Tap inTap 	= new Hfs(new TextLine(), inputPath); //input
        Tap outTap  = new Hfs(new TextDelimited(true, ";"), outputPath, SinkMode.REPLACE); //output
        //Tap top10Tap = new Hfs( new TextDelimited(true, ";"), sortedTopTen, SinkMode.REPLACE);
        
        // Parse the line of input and break them into five fields
        RegexParser parser = new RegexParser(new Fields("ip", "time", "request", "response", "size"), 
        		"^([^ ]*) \\S+ \\S+ \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3}) ([^ ]*).*$", new int[]{1, 2, 3, 4, 5});
        
        // "Each" pipe applies a Function or Filter Operation to each Tuple that passes through it.
        Pipe processPipe = new Each("processPipe", new Fields("line"), parser, Fields.RESULTS);
        
        // Grouping by 'ip' field: 
        // "GroupBy" manages one input Tuple stream and, groups the stream on selected fields in the tuple stream. 
        processPipe = new GroupBy(processPipe, new Fields("ip"));
        
        // Aggregate each "ip" group using Count function:
        // "Every" pipe applies an Aggregator (like count, or sum) or Buffer (a sliding window) Operation to every group of Tuples that pass through it.
        processPipe = new Every(processPipe, Fields.GROUP, new Count(new Fields("IPcount")), Fields.ALL);
        
        // After aggregation counter for each "ip," sort the counts. "true" is descending order
        Pipe sortedCountByIpPipe = new GroupBy(processPipe, new Fields("IPcount"), true);
        
        // Limit them to the first 10, in the descending order
        sortedCountByIpPipe = new Each(sortedCountByIpPipe, new Fields("IPcount"), new Limit(10));
        
        // Join the pipe together in the flow, creating inputs and outputs (taps)
        FlowDef flowDef = FlowDef.flowDef()
    		   .addSource(processPipe, inTap)
    		   //.addTailSink(processPipe, outTap) // comment to use sorted top 10
    		   //.addTailSink(sortedCountByIpPipe, top10Tap) //uncomment to use sorted top 10
    		   .addTailSink(sortedCountByIpPipe, outTap) //uncomment to use sorted top 10
    		   .setName("DataProcessing");
        Properties properties = AppProps.appProps()
        		.setName("DataProcessing")
        		.buildProperties();
        Flow parsedLogFlow = new HadoopFlowConnector(properties).connect(flowDef);
        
        //Finally, execute the flow.
        parsedLogFlow.complete();
    }
}

@@ -1748,23 +1748,21 @@
    Pipe 
-process
+top10
 Pipe = n
@@ -1770,23 +1770,21 @@
 w Each(%22
-process
+top10
 Pipe%22, n
@@ -1988,39 +1988,37 @@
 tream. %0A        
-process
+top10
 Pipe = new Group
@@ -2012,39 +2012,37 @@
 e = new GroupBy(
-process
+top10
 Pipe, new Fields
@@ -2279,23 +2279,21 @@
         
-process
+top10
 Pipe = n
@@ -2301,23 +2301,21 @@
 w Every(
-process
+top10
 Pipe, Fi
@@ -2491,31 +2491,26 @@
    Pipe 
-sorted
+top10
 Count
-ByIp
 Pipe = n
@@ -2520,23 +2520,21 @@
 GroupBy(
-process
+top10
 Pipe, ne
@@ -2641,31 +2641,26 @@
         
-sorted
+top10
 Count
-ByIp
 Pipe = n
@@ -2667,31 +2667,26 @@
 ew Each(
-sorted
+top10
 Count
-ByIp
 Pipe, ne
@@ -2921,23 +2921,21 @@
 ailSink(
-process
+top10
 Pipe, ou
@@ -2992,39 +2992,34 @@
 addTailSink(
-sorted
+top10
 Count
-ByIp
 Pipe, top10T
@@ -3081,23 +3081,18 @@
 ink(
-sorted
+top10
 Count
-ByIp
 Pipe
@@ -3153,30 +3153,23 @@
 etName(%22
-DataProcessing
+Top10IP
 %22);%0A    
@@ -3240,22 +3240,15 @@
 me(%22
-DataProcessing
+Top10IP
 %22)%0A 

Update Main.java