package com.amazonaws.vivanih.hadoop.cascading;

import java.util.Properties;

import cascading.flow.Flow;
import cascading.flow.FlowConnector;
import cascading.flow.FlowDef;
import cascading.flow.hadoop.HadoopFlowConnector;
import cascading.operation.aggregator.Count;
import cascading.operation.regex.RegexSplitGenerator;
import cascading.pipe.Each;
import cascading.pipe.Every;
import cascading.pipe.GroupBy;
import cascading.pipe.Pipe;
import cascading.property.AppProps;
import cascading.scheme.Scheme;
import cascading.scheme.hadoop.TextDelimited;
import cascading.tap.Tap;
import cascading.tap.hadoop.Hfs;
import cascading.tuple.Fields;


public class
  Main
  {
  public static void
  main( String[] args )
    {
    String inPath = args[ 0 ];
    String outPath = args[ 1 ];

    Properties properties = new Properties();
    AppProps.setApplicationJarClass( properties, Main.class );
    FlowConnector flowConnector = new HadoopFlowConnector( properties );

    //scheme definition:
    Scheme inScheme = new TextDelimited( new Fields("LoadId", "MRP", "ServicerName", "CIR", "UPB", "LoanAge", "RMLM" , "ARMM", "MadurityDate", "MSA", "CLDS", "ModificationFlag", "ZBC", "ZBED", "RepurchaseIndicator"), "|");
    Scheme outScheme = new TextDelimited( new Fields( "MRP", "UPB" ) );

    // create source and sink taps
    Tap inTap = new Hfs( inScheme, inPath );
    Tap outTap = new Hfs( outScheme, outPath );


    // specify a pipe to connect the taps
    Pipe copyPipe = new Pipe( "copy" );
    // connect the taps, pipes, etc., into a flow
    FlowDef flowDef = FlowDef.flowDef()
    .addSource( copyPipe, inTap )
    .addTailSink( copyPipe, outTap );
    // run the flow
    flowConnector.connect( flowDef ).complete();
    }
  }

@@ -70,16 +70,43 @@
 perties;
+%0Aimport java.util.Calendar;
 %0A%0Aimport
@@ -290,16 +290,63 @@
 .Count;%0A
+import cascading.operation.aggregator.Average;%0A
 import c
@@ -387,24 +387,99 @@
 tGenerator;%0A
+import cascading.operation.text.DateParser;%0Aimport cascading.tap.SinkMode;%0A
 import casca
@@ -788,16 +788,17 @@
 elds;%0A%0A%0A
+%0A
 public c
@@ -1359,24 +1359,26 @@
 , %22%7C%22);%0A    
+//
 Scheme outSc
@@ -1514,24 +1514,26 @@
 Path );%0A    
+//
 Tap outTap =
@@ -1568,92 +1568,977 @@
  );%0A
-%0A%0A    // specify a pipe to connect the taps%0A    Pipe copyPipe = new Pipe( %22copy%22
+    Tap outTap = new Hfs( new TextDelimited(true, %22;%22), outPath );%0A    //Tap averageTap = new Hfs( outSchemeAverage, averagePath );%0A%0A    //Parsing the date:%0A    DateParser dateParser= new DateParser(new Fields(%22month%22, %22day%22, %22year%22), new int%5B%5D %7B Calendar.MONTH, Calendar.DAY_OF_MONTH, Calendar.YEAR %7D, %22MM/dd/yyyy%22);%0A    //DateParser dateParser= new DateParser(new Fields(%22month%22, %22day%22, %22year%22), %22MM/dd/yyyy%22);%0A    //parse pipe will parse the MRP field into month, day, year. /with Fields.ALL I get original fields + new fields. With Fields.RESULTS, I get only the date parsed.%0A    Pipe parsePipe = new Each(%22parsePipe%22, new Fields(%22MRP%22), dateParser , Fields.ALL);%0A    // aggregate by year, month. input is previous parsePipe%0A    Pipe averagePipe = new GroupBy( %22averagePipe%22, parsePipe, new Fields(%22year%22, %22month%22 ));%0A    // average each aggregation%0A    averagePipe = new Every(averagePipe, new Fields(%22UPB%22), new Average(new Fields(%22UPBaverage%22)), Fields.ALL
  );%0A
+%0A%0A
     
@@ -2639,20 +2639,23 @@
 Source( 
-copy
+average
 Pipe, in
@@ -2681,13 +2681,15 @@
 ink(
- copy
+average
 Pipe

Update Main.java