@@ -0,0 +1,2841 @@
+package com.amazonaws.vivanih.hadoop.cascading;%0A%0Aimport java.util.Properties;%0A%0Aimport cascading.flow.Flow;%0Aimport cascading.flow.hadoop.HadoopFlowConnector;%0Aimport cascading.operation.regex.RegexParser;%0Aimport cascading.pipe.Each;%0Aimport cascading.pipe.Pipe;%0Aimport cascading.property.AppProps;%0Aimport cascading.scheme.hadoop.TextLine;%0Aimport cascading.tap.SinkMode;%0Aimport cascading.tap.Tap;%0Aimport cascading.tap.hadoop.Hfs;%0Aimport cascading.tap.hadoop.Lfs;%0Aimport cascading.tuple.Fields;%0A%0A/**%0A *%0A */%0Apublic class LogParser%0A  %7B%0A  public static void main( String%5B%5D args )%0A    %7B%0A    String inputPath = args%5B 0 %5D;%0A    String outputPath = args%5B 1 %5D;%0A%0A    // define what the input file looks like, %22offset%22 is bytes from beginning%0A    TextLine scheme = new TextLine( new Fields( %22offset%22, %22line%22 ) );%0A%0A    // create SOURCE tap to read a resource from the local file system, if input is not an URL%0A    Tap logTap = inputPath.matches( %22%5E%5B%5E:%5D+://.*%22 ) ? new Hfs( scheme, inputPath ) : new Lfs( scheme, inputPath );%0A%0A    // create an assembly to parse an Apache log file and store on an HDFS cluster%0A%0A    // declare the field names we will parse out of the log file%0A    Fields apacheFields = new Fields( %22ip%22, %22time%22, %22method%22, %22event%22, %22status%22, %22size%22 );%0A%0A    // define the regular expression to parse the log file with%0A    String apacheRegex = %22%5E(%5B%5E %5D*) +%5B%5E %5D* +%5B%5E %5D* +%5C%5C%5B(%5B%5E%5D%5D*)%5C%5C%5D +%5C%5C%5C%22(%5B%5E %5D*) (%5B%5E %5D*) %5B%5E %5D*%5C%5C%5C%22 (%5B%5E %5D*) (%5B%5E %5D*).*$%22;%0A%0A    // declare the groups from the above regex we want to keep. each regex group will be given%0A    // a field name from 'apacheFields', above, respectively%0A    int%5B%5D allGroups = %7B1, 2, 3, 4, 5, 6%7D;%0A%0A    // create the parser%0A    RegexParser parser = new RegexParser( apacheFields, apacheRegex, allGroups );%0A%0A    // create the import pipe element, with the name 'import', and with the input argument named %22line%22%0A    // replace the incoming tuple with the parser results%0A    // %22line%22 -%3E parser -%3E %22ts%22%0A    Pipe importPipe = new Each( %22import%22, new Fields( %22line%22 ), parser, Fields.RESULTS );%0A%0A    // create a SINK tap to write to the default filesystem%0A    // by default, TextLine writes all fields out%0A    Tap remoteLogTap = new Hfs( new TextLine(), outputPath, SinkMode.REPLACE );%0A%0A    // set the current job jar%0A    Properties properties = new Properties();%0A    AppProps.setApplicationJarClass( properties, Main.class );%0A%0A    // connect the assembly to the SOURCE and SINK taps%0A    Flow parsedLogFlow = new HadoopFlowConnector( properties ).connect( logTap, remoteLogTap, importPipe );%0A%0A    // optionally print out the parsedLogFlow to a DOT file for import into a graphics package%0A    // parsedLogFlow.writeDOT( %22logparser.dot%22 );%0A%0A    // start execution of the flow (either locally or on the cluster%0A    parsedLogFlow.start();%0A%0A    // block until the flow completes%0A    parsedLogFlow.complete();%0A    %7D%0A  %7D%0A

Create logparser.java