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.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 );

    // create source and sink taps
    Tap inTap = new Hfs( new TextDelimited( false, "|" ), inPath );
    Tap outTap = new Hfs( new TextDelimited( false, "|" ), outPath );

    //define the fields:
    Fields recordFields = new Fields( "LoadId", "MRP", "ServicerName", "CIR", "UPB", "LoanAge", "RMLM" , "ARMM", "MadurityDate", "MSA", "CLDS", "ModificationFlag", "ZBC", "ZBED", "RepurchaseIndicator");
    
    // 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();
    }
  }

@@ -469,16 +469,48 @@
 pProps;%0A
+import cascading.scheme.Scheme;%0A
 import c
@@ -971,226 +971,67 @@
   //
- create source and sink taps%0A    Tap inTap = new Hfs( new TextDelimited( false, %22%7C%22 ), inPath );%0A    Tap outTap = new Hfs( new TextDelimited( false, %22%7C%22 ), outPath );%0A%0A    //define the fields:%0A    Fields recordFields =
+scheme definition:%0A    Scheme inScheme = new TextDelimited(
  new
@@ -1038,17 +1038,16 @@
  Fields(
- 
 %22LoadId%22
@@ -1201,22 +1201,226 @@
 icator%22)
-;%0A    
+, %22%7C%22);%0A    Scheme outScheme = new TextDelimited( new Fields( %22MRP%22, %22UPB%22 ) );%0A%0A    // create source and sink taps%0A    Tap inTap = new Hfs( inScheme, inPath );%0A    Tap outTap = new Hfs( outScheme, outPath );%0A%0A
 %0A    // 

Update Main.java