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.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.tap.Tap;
import cascading.tap.hadoop.Hfs;
import cascading.tap.hadoop.Lfs;
import cascading.tuple.Fields;
import org.elasticsearch.hadoop.cascading.EsTap;


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

    // create the source tap
    //Tap inTap = new Hfs( new TextDelimited( true, "\t" ), inPath );
    Tap inTap = new Lfs(new TextDelimited(new Fields("id", "name", "url", "picture")), "/mnt/data/");

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

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

@@ -72,16 +72,66 @@
 rties;%0A%0A
+import org.elasticsearch.hadoop.cascading.EsTap;%0A%0A
 import c
@@ -948,16 +948,57 @@
 imited;%0A
+import cascading.scheme.hadoop.TextLine;%0A
 import c
@@ -1116,56 +1116,8 @@
 ds;%0A
-import org.elasticsearch.hadoop.cascading.EsTap;
 %0A%0A%0Ap
@@ -1221,24 +1221,26 @@
 s%5B 0 %5D;%0A    
+//
 String outPa
@@ -1365,16 +1365,346 @@
 lass );%0A
+    //we tell cascading we are using JSON files as input:%0A    properties.setProperty(%22es.input.json%22, %22true%22);%0A    //disable speculative execution%0A    properties.setProperty(%22mapreduce.map.speculative%22, %22false%22);%0A    properties.setProperty(%22mapreduce.reduce.speculative%22, %22false%22);%0A    properties.setProperty(%22es.port%22, %229202%22);%0A%0A
    // Ap
@@ -2023,16 +2023,95 @@
 rties );
+%0A    //FlowConnector flowConnector = new Hadoop2MR1FlowConnector( properties );
 %0A%0A    //
@@ -2128,31 +2128,44 @@
 e source tap
+ for JSON input
 %0A    
-//
 Tap inTap = 
@@ -2181,38 +2181,39 @@
 new Text
-Delimited( true, %22%5Ct%22 
+Line(new Fields(%22line%22)
 ), inPat
@@ -2217,24 +2217,26 @@
 Path );%0A    
+//
 Tap inTap = 
@@ -2310,20 +2310,14 @@
 )), 
-%22/mnt/data/%22
+inPath
 );%0A%0A
@@ -2461,48 +2461,8 @@
 sts%22
- , new Fields(%22name%22, %22url%22, %22picture%22) 
 );%0A%0A
@@ -2791,12 +2791,19 @@
 ;%0A    %7D%0A  %7D%0A
+      %0A

Update Main.java