package com.amazonaws.vivanih.hadoop.cascading;

import java.util.Properties;

import org.elasticsearch.hadoop.cascading.EsTap;

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.regex.RegexGenerator;
import cascading.operation.regex.RegexFilter;
import cascading.operation.text.DateParser;
import cascading.operation.text.DateFormatter;
import cascading.operation.text.FieldFormatter;
import cascading.tap.SinkMode;
import cascading.pipe.Each;
import cascading.pipe.Every;
import cascading.pipe.GroupBy;
import cascading.pipe.Pipe;
import cascading.pipe.assembly.Unique;
import cascading.property.AppProps;
import cascading.scheme.Scheme;
import cascading.scheme.hadoop.TextDelimited;
import cascading.scheme.hadoop.TextLine;
import cascading.tap.Tap;
import cascading.tap.hadoop.Hfs;
import cascading.tap.hadoop.Lfs;
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 );
    //we tell cascading we are using JSON files as input:
    properties.setProperty("es.input.json", "true");
    //disable speculative execution
    properties.setProperty("mapreduce.map.speculative", "false");
    properties.setProperty("mapreduce.reduce.speculative", "false");
    properties.setProperty("es.port", "9202");

   // AppProps.setApplicationJarPath(properties, "/home/hadoop/impatient.jar");
    //FlowConnector flowConnector = new Hadoop2MR1FlowConnector( properties );
    //test elasticsearch set property
    //props.setProperty("es.index.auto.create", "false");

    FlowConnector flowConnector = new HadoopFlowConnector( properties );
    //FlowConnector flowConnector = new Hadoop2MR1FlowConnector( properties );

    // create the source tap for JSON input
    Tap inTap = new Hfs( new TextLine(new Fields("line")), inPath );
    //Tap inTap = new Lfs(new TextDelimited(new Fields("id", "name", "url", "picture")), inPath);

    // create the sink tap
    //Tap outTap = new Hfs( new TextDelimited( true, "\t" ), outPath );
    Tap outTap = new EsTap("radio/artists");

    // specify a pipe to connect the taps
    Pipe copyPipe = new Pipe( "write-to-Es" );

    // 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();
    }
  }
      

@@ -1221,16 +1221,48 @@
 s%5B 0 %5D;%0A
+    String outPath = args%5B 1 %5D;%0A
     //St
@@ -1262,27 +1262,28 @@
    //String 
-out
+trap
 Path = args%5B
@@ -2246,24 +2246,99 @@
 , inPath );%0A
+    //Tap trapTap = new Hfs( new TextLine(new Fields(%22line%22)), trapPath );%0A
     //Tap in
@@ -2491,30 +2491,31 @@
 Text
-Delimited( true, %22%5Ct%22 
+Line(new Fields(%22line%22)
 ), o
@@ -2566,24 +2566,248 @@
 artists%22);%0A%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    parsePipe = new Each( parsePipe, new Fields( %22line%22 ), splitter, Fields.RESULTS );%0A%0A
     // speci
@@ -2836,24 +2836,26 @@
 he taps%0A    
+//
 Pipe copyPip
@@ -2992,16 +2992,52 @@
 dSource(
+ parsePipe, inTap )%0A     //.addTrap(
  copyPip
@@ -3035,26 +3035,28 @@
 ( copyPipe, 
-in
+trap
 Tap )%0A     .
@@ -3068,20 +3068,21 @@
 ilSink( 
-copy
+parse
 Pipe, ou

Update Main.java