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 docPath = args[ 0 ];
    String wcPath = args[ 1 ];

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

    // create source and sink taps
    Tap docTap = new Hfs( new TextDelimited( true, "|" ), docPath );
    Tap wcTap = new Hfs( new TextDelimited( true, "|" ), wcPath );

    // specify a regex operation to split the "document" text lines into a token stream
    Fields token = new Fields( "token" );
    Fields text = new Fields( "text" );
    RegexSplitGenerator splitter = new RegexSplitGenerator( token, "[ \\[\\]\\(\\),.]" );
    // only returns "token"
    Pipe docPipe = new Each( "token", text, splitter, Fields.RESULTS );

    // determine the word counts
    Pipe wcPipe = new Pipe( "wc", docPipe );
    wcPipe = new GroupBy( wcPipe, token );
    wcPipe = new Every( wcPipe, Fields.ALL, new Count(), Fields.ALL );

    // connect the taps, pipes, etc., into a flow
    FlowDef flowDef = FlowDef.flowDef()
     .setName( "wc" )
     .addSource( docPipe, docTap )
     .addTailSink( wcPipe, wcTap );

    // write a DOT file and run the flow
    Flow wcFlow = flowConnector.connect( flowDef );
    wcFlow.writeDOT( "dot/wc.dot" );
    wcFlow.complete();
    }
  }

@@ -693,19 +693,18 @@
  String 
-doc
+in
 Path = a
@@ -724,18 +724,19 @@
  String 
-wc
+out
 Path = a
@@ -972,19 +972,18 @@
     Tap 
-doc
+in
 Tap = ne
@@ -1021,19 +1021,18 @@
  %22%7C%22 ), 
-doc
+in
 Path );%0A
@@ -1039,18 +1039,19 @@
     Tap 
-wc
+out
 Tap = ne
@@ -1089,18 +1089,19 @@
  %22%7C%22 ), 
-wc
+out
 Path );%0A
@@ -1097,24 +1097,29 @@
 outPath );%0A%0A
+    %0A
     // speci
@@ -1127,544 +1127,72 @@
 y a 
-regex operation to split the %22document%22 text lines into a token stream%0A    Fields token = new Fields( %22token%22 );%0A    Fields text = new Fields( %22text%22 );%0A    RegexSplitGenerator splitter = new RegexSplitGenerator( token, %22%5B %5C%5C%5B%5C%5C%5D%5C%5C(%5C%5C),.%5D%22 );%0A    // only returns %22token%22%0A    Pipe docPipe = new Each( %22token%22, text, splitter, Fields.RESULTS );%0A%0A    // determine the word counts%0A    Pipe wcPipe = new Pipe( %22wc%22, docPipe );%0A    wcPipe = new GroupBy( wcPipe, token );%0A    wcPipe = new Every( wcPipe, Fields.ALL, new Count(), Fields.ALL );%0A
+pipe to connect the taps%0A    Pipe copyPipe = new Pipe( %22copy%22 );
 %0A   
@@ -1282,31 +1282,8 @@
 f()%0A
-     .setName( %22wc%22 )%0A 
     
@@ -1298,27 +1298,26 @@
 ce( 
-doc
+copy
 Pipe, 
-doc
+in
 Tap )%0A
- 
     
@@ -1334,53 +1334,34 @@
 nk( 
-w
 c
+opy
 Pipe, 
-wc
+out
 Tap );%0A
-%0A
     //
- write a DOT file and
  run
@@ -1377,22 +1377,8 @@
 %0A   
- Flow wcFlow =
  flo
@@ -1410,57 +1410,8 @@
 ef )
-;%0A    wcFlow.writeDOT( %22dot/wc.dot%22 );%0A    wcFlow
 .com

Update Main.java