package com.amazonaws.vivanih.hadoop.cascading;

import java.util.Properties;
import java.util.Calendar;

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.aggregator.Average;
import cascading.operation.regex.RegexSplitGenerator;
import cascading.operation.text.DateParser;
import cascading.tap.SinkMode;
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 );
    Tap outTap = new Hfs( new TextDelimited(true, ";"), outPath );
    //Tap averageTap = new Hfs( outSchemeAverage, averagePath );

    //Parsing the date:
    DateParser dateParser= new DateParser(new Fields("month", "day", "year"), new int[] { Calendar.MONTH, Calendar.DAY_OF_MONTH, Calendar.YEAR }, "MM/dd/yyyy");
    //DateParser dateParser= new DateParser(new Fields("month", "day", "year"), "MM/dd/yyyy");
    //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.
    Pipe parsePipe = new Each("parsePipe", new Fields("MRP"), dateParser , Fields.ALL);
    // aggregate by year, month. input is previous parsePipe
    Pipe averagePipe = new GroupBy( "averagePipe", parsePipe, new Fields("year", "month" ));
    // average each aggregation
    averagePipe = new Every(averagePipe, new Fields("UPB"), new Average(new Fields("UPBaverage")), Fields.ALL );


    // connect the taps, pipes, etc., into a flow
    FlowDef flowDef = FlowDef.flowDef()
    .addSource( averagePipe, inTap )
    .addTailSink(averagePipe, outTap );
    // run the flow
    flowConnector.connect( flowDef ).complete();
    }
  }

@@ -97,16 +97,51 @@
 alendar;
+%0Aimport java.text.SimpleDateFormat;
 %0A%0Aimport
@@ -617,16 +617,55 @@
 e.Pipe;%0A
+import cascading.pipe.assembly.Unique;%0A
 import c
@@ -861,16 +861,17 @@
 ields;%0A%0A
+%0A
 %0A%0Apublic
@@ -1438,18 +1438,16 @@
 %22);%0A    
-//
 Scheme o
@@ -1493,23 +1493,45 @@
 s( %22
-MRP%22, %22UPB
+year%22, %22month%22, %22UPBaverage%22),%22%5Ct
 %22 ) 
-)
 ;%0A%0A 
@@ -1609,26 +1609,24 @@
 Path );%0A    
-//
 Tap outTap =
@@ -1653,32 +1653,34 @@
  outPath );%0A    
+//
 Tap outTap = new
@@ -1710,17 +1710,18 @@
 (true, %22
-;
+%5Ct
 %22), outP
@@ -1730,73 +1730,8 @@
 h );
-%0A    //Tap averageTap = new Hfs( outSchemeAverage, averagePath );
 %0A%0A  
@@ -1750,17 +1750,32 @@
 the date
-:
+. (0 is January)
 %0A    Dat
@@ -2278,16 +2278,280 @@
 s.ALL);%0A
+%0A    //Pipe to filter duplicates. It is like select distinct. It is like using combiners.%0A    //http://docs.cascading.org/cascading/2.2/javadoc/cascading/pipe/assembly/Unique.html%0A    Pipe uniquePipe = new Unique(%22unique%22, parsePipe, new Fields(%22LoadId%22,%22MRP%22));%0A%0A
     // a
@@ -2650,28 +2650,29 @@
 eragePipe%22, 
-pars
+uniqu
 ePipe, new F
@@ -2979,27 +2979,104 @@
     
-.addTailSink(averag
+//.addSource( parsePipe, inTap )%0A    .addTailSink(averagePipe, outTap );%0A    //.addTailSink(pars
 ePip

Update Main.java