@@ -0,0 +1,2605 @@
+import cascading.flow.Flow;%0Aimport cascading.flow.FlowDef;%0Aimport cascading.flow.hadoop.HadoopFlowConnector;%0Aimport cascading.operation.aggregator.Count;%0Aimport cascading.operation.filter.Sample;%0Aimport cascading.operation.filter.Limit;%0Aimport cascading.operation.regex.RegexParser;%0Aimport cascading.operation.text.DateParser;%0Aimport cascading.pipe.*;%0Aimport cascading.property.AppProps;%0Aimport cascading.scheme.hadoop.TextDelimited;%0Aimport cascading.scheme.hadoop.TextLine;%0Aimport cascading.tap.SinkMode;%0Aimport cascading.tap.Tap;%0Aimport cascading.tap.hadoop.Hfs;%0Aimport cascading.tuple.Fields;%0Aimport java.util.Properties;%0A %0Apublic class Main %7B%0A    public static void main(String%5B%5D args) %7B%0A %0A    %09// input (taps) and output (sinks)%0A        String inputPath %09%09= args%5B0%5D;%0A        String outputPath %09= args%5B1%5D;%0A        // sources and sinks%0A        Tap inTap %09= new Hfs(new TextLine(), inputPath);%0A        Tap outTap  = new Hfs(new TextDelimited(true, %22%5Ct%22), outputPath, SinkMode.REPLACE);%0A        // Parse the line of input and break them into five fields%0A        RegexParser parser = new RegexParser(new Fields(%22ip%22, %22time%22, %22request%22, %22response%22, %22size%22), %0A        %09%09%22%5E(%5B%5E %5D*) %5C%5CS+ %5C%5CS+ %5C%5C%5B(%5B%5C%5Cw:/%5D+%5C%5Cs%5B+%5C%5C-%5D%5C%5Cd%7B4%7D)%5C%5C%5D %5C%22(.+?)%5C%22 (%5C%5Cd%7B3%7D) (%5B%5E %5D*).*$%22, new int%5B%5D%7B1, 2, 3, 4, 5%7D);%0A        // Create a pipe for processing each line at a time%0A        Pipe processPipe = new Each(%22processPipe%22, new Fields(%22line%22), parser, Fields.RESULTS);%0A        // Group the stream within the pipe by the field %22ip%22%0A        processPipe = new GroupBy(processPipe, new Fields(%22ip%22));%0A        // Aggregate each %22ip%22 group using the Cascading built in Count function%0A        processPipe = new Every(processPipe, Fields.GROUP, new Count(new Fields(%22IPcount%22)), Fields.ALL);%0A        // After aggregation counter for each %22ip,%22 sort the counts%0A        Pipe sortedCountByIpPipe = new GroupBy(processPipe, new Fields(%22IPcount%22), true);%0A        // Limit them to the first 10, in the descending order%0A        sortedCountByIpPipe = new Each(sortedCountByIpPipe, new Fields(%22IPcount%22), new Limit(10));%0A        // Join the pipe together in the flow, creating inputs and outputs (taps)%0A        FlowDef flowDef = FlowDef.flowDef()%0A    %09%09   .addSource(processPipe, inTap)%0A    %09%09   .addTailSink(sortedCountByIpPipe, outTap)%0A    %09%09   .setName(%22DataProcessing%22);%0A        Properties properties = AppProps.appProps()%0A        %09%09.setName(%22DataProcessing%22)%0A        %09%09.buildProperties();%0A        Flow parsedLogFlow = new HadoopFlowConnector(properties).connect(flowDef);%0A        //Finally, execute the flow.%0A        parsedLogFlow.complete();%0A    %7D%0A%7D%0A

Create Main2.java