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) {
 
    	// input (taps) and output (sinks)
        String inputPath 		= args[0];
        String outputPath 	= args[1];
        // sources and sinks
        Tap inTap 	= new Hfs(new TextLine(), inputPath);
        Tap outTap  = new Hfs(new TextDelimited(true, "\t"), outputPath, 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});
        // Create a pipe for processing each line at a time
        Pipe processPipe = new Each("processPipe", new Fields("line"), parser, Fields.RESULTS);
        // Group the stream within the pipe by the field "ip"
        processPipe = new GroupBy(processPipe, new Fields("ip"));
        // Aggregate each "ip" group using the Cascading built in Count function
        processPipe = new Every(processPipe, Fields.GROUP, new Count(new Fields("IPcount")), Fields.ALL);
        // After aggregation counter for each "ip," sort the counts
        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(sortedCountByIpPipe, outTap)
    		   .setName("DataProcessing");
        Properties properties = AppProps.appProps()
        		.setName("DataProcessing")
        		.buildProperties();
        Flow parsedLogFlow = new HadoopFlowConnector(properties).connect(flowDef);
        //Finally, execute the flow.
        parsedLogFlow.complete();
    }
}

@@ -803,16 +803,25 @@
 rgs%5B1%5D;%0A
+        %0A
         
@@ -841,16 +841,16 @@
 d sinks%0A
-
         
@@ -957,10 +957,9 @@
 e, %22
-%5Ct
+;
 %22), 
@@ -989,16 +989,25 @@
 PLACE);%0A
+        %0A
         
@@ -1289,16 +1289,25 @@
 4, 5%7D);%0A
+        %0A
         
@@ -1454,16 +1454,25 @@
 SULTS);%0A
+        %0A
         
@@ -1587,24 +1587,33 @@
 lds(%22ip%22));%0A
+        %0A
         // A
@@ -1787,16 +1787,25 @@
 s.ALL);%0A
+        %0A
         
@@ -1954,16 +1954,25 @@
  true);%0A
+        %0A
         
@@ -2121,24 +2121,33 @@
 Limit(10));%0A
+        %0A
         // J
@@ -2503,24 +2503,24 @@
 operties();%0A
-
         Flow
@@ -2590,16 +2590,25 @@
 owDef);%0A
+        %0A
         

Update Main2.java