@@ -0,0 +1,3267 @@
+package com.amazonaws.vivanih.hadoop.cascading;%0A%0Aimport java.util.Properties;%0A%0Aimport org.elasticsearch.hadoop.cascading.EsTap;%0A%0Aimport cascading.flow.Flow;%0Aimport cascading.flow.FlowConnector;%0Aimport cascading.flow.FlowDef;%0Aimport cascading.flow.hadoop.HadoopFlowConnector;%0Aimport cascading.operation.aggregator.Count;%0Aimport cascading.operation.aggregator.Average;%0Aimport cascading.operation.regex.RegexSplitGenerator;%0Aimport cascading.operation.regex.RegexGenerator;%0Aimport cascading.operation.regex.RegexFilter;%0Aimport cascading.operation.text.DateParser;%0Aimport cascading.operation.text.DateFormatter;%0Aimport cascading.operation.text.FieldFormatter;%0Aimport cascading.tap.SinkMode;%0Aimport cascading.pipe.Each;%0Aimport cascading.pipe.Every;%0Aimport cascading.pipe.GroupBy;%0Aimport cascading.pipe.Pipe;%0Aimport cascading.pipe.assembly.Unique;%0Aimport cascading.property.AppProps;%0Aimport cascading.scheme.Scheme;%0Aimport cascading.scheme.hadoop.TextDelimited;%0Aimport cascading.scheme.hadoop.TextLine;%0Aimport cascading.tap.Tap;%0Aimport cascading.tap.hadoop.Hfs;%0Aimport cascading.tap.hadoop.Lfs;%0Aimport cascading.tuple.Fields;%0A%0A%0A%0Apublic class%0A  Main%0A  %7B%0A  public static void%0A  main( String%5B%5D args )%0A    %7B%0A    String inPath = args%5B 0 %5D;%0A%0A    Properties properties = new Properties();%0A    AppProps.setApplicationJarClass( properties, Main.class );%0A    //we tell cascading we are using JSON files as input:%0A    //As this is just used when calling EsTap library, we can apply some filters before output is indexed on Elasticsearch%0A    properties.setProperty(%22es.input.json%22, %22true%22);%0A    //disable speculative execution:  will schedule redundant copies of the remaining tasks across several nodes which do not have other work to perform.%0A    properties.setProperty(%22mapreduce.map.speculative%22, %22false%22);%0A    properties.setProperty(%22mapreduce.reduce.speculative%22, %22false%22);%0A    //elasticsearch bootstrapaction uses port 9202 on the slave nodes. As 9200 is used for HDFS internode communication.%0A    properties.setProperty(%22es.port%22, %229202%22);%0A%0A    //by default index will be auto created%0A    //propperties.setProperty(%22es.index.auto.create%22, %22false%22);%0A%0A    FlowConnector flowConnector = new HadoopFlowConnector( properties );%0A%0A    // create the source tap for JSON input%0A    Tap inTap = new Hfs( new TextLine(new Fields(%22line%22)), inPath );%0A%0A    // create the sink tap%0A    // Output is sent directly to Elasticsearch%0A    Tap outTap = new EsTap(%22common-crawl/wat%22);%0A    //if we want to write to HDFS instead of elasticsearch:%0A    //Tap outTap = new Hfs( new TextLine(new Fields(%22line%22)), outPath );%0A%0A    //The input files are parsed using Regex to remove WARC Headers and process a clean JSON file as input%0A    Pipe parsePipe = new Pipe( %22parsePipe%22 );%0A    RegexGenerator splitter=new RegexGenerator(new Fields(%22json%22),%22%5E%5C%5C%7B%5C%22Envelope%5C%22.*$%22);%0A    //Each  operator will filter each entry in the Tuple stream depending on the Regex definition:%0A    parsePipe = new Each( parsePipe, new Fields( %22line%22 ), splitter, Fields.RESULTS );%0A%0A    // connect the taps, pipes, etc., into a flow%0A    FlowDef flowDef = FlowDef.flowDef()%0A     .addSource( parsePipe, inTap )%0A     .addTailSink( parsePipe, outTap );%0A%0A    // run the flow%0A    flowConnector.connect( flowDef ).complete();%0A    %7D%0A  %7D%0A      %0A

Create Main.java