package com.thinkbiganalytics.esbulkloader;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ElasticsearchBulkLoader {
    
    private static Log log = LogFactory.getLog(ElasticsearchBulkLoader.class);
    public static void main(String args[]) throws Exception {
	
	log.info(String.format("%s %s %s", args[0], args[1], args[2]));
	
	Configuration conf = new Configuration();

	Job job = new Job(conf, args[0]);
	job.setJarByClass(ElasticsearchBulkLoader.class);
	job.setMapperClass(IndexFilesMapper.class);
	job.setNumReduceTasks(0);
	job.setMapOutputKeyClass(NullWritable.class);
	job.setMapOutputValueClass(Text.class);
	job.setInputFormatClass(TextInputFormat.class);
	job.setOutputFormatClass(ElasticsearchBulkFormat.class);
	
	 FileInputFormat.addInputPath(job, new Path(args[1]));
	 FileOutputFormat.setOutputPath(job, new Path(args[2]));

	// actually run the job
	boolean success = job.waitForCompletion(true);

	System.exit(success ? 0 : 1);
    }

    public class IndexFilesMapper extends
	    Mapper<NullWritable, Text, NullWritable, Text> {

	@Override
	public void setup(Context context) throws IOException {

	}

	@Override
	public void map(NullWritable key, Text value, Context context)
		throws IOException {
	    // parse the binary, convert it to JSON, and index that JSON
	    try {
		String id="123";
		String bulkJsonFormat = "{\"index\":{ \"_index\":\"tv_perf_v3\", \"_type\":\"voters\",\"_id\":\"%s\" } }\n%s\n";

		context.write(null, new Text(String.format(bulkJsonFormat, id, value)));

	    } catch (Exception e) {
		throw new IOException(e);
	    }
	}

    }

}

@@ -42,36 +42,8 @@
 r;%0A%0A
-import java.io.IOException;%0A
 %0Aimp
@@ -169,16 +169,58 @@
 ration;%0A
+import org.apache.hadoop.conf.Configured;%0A
 import o
@@ -365,51 +365,8 @@
 ob;%0A
-import org.apache.hadoop.mapreduce.Mapper;%0A
 impo
@@ -548,16 +548,94 @@
 tFormat;
+%0Aimport org.apache.hadoop.util.Tool;%0Aimport org.apache.hadoop.util.ToolRunner;
 %0A%0Apublic
@@ -669,12 +669,45 @@
 der 
+extends Configured implements Tool 
 %7B%0A
-  
   %0A 
@@ -784,16 +784,17 @@
 class);%0A
+%0A
     publ
@@ -823,15 +823,15 @@
 ring
+%5B%5D
  args
-%5B%5D
 ) th
@@ -852,18 +852,231 @@
  %7B%0A%09
-%0A%09log.info
+// Use generic options tool runner%0A%09int exitCode = ToolRunner.run(new ElasticsearchBulkLoader(), args); %0A%09System.exit(exitCode); %0A    %7D%0A%0A    @Override%0A    public int run(String%5B%5D args) throws Exception %7B%0A%09System.out.println
 (Str
@@ -1279,26 +1279,40 @@
 erClass(
-IndexFiles
+ElasticsearchBulkIndexer
 Mapper.c
@@ -1734,670 +1734,34 @@
 e);%0A
-%0A%09System.exit(success ? 0 : 1);%0A    %7D%0A%0A    public class IndexFilesMapper extends%0A%09    Mapper%3CNullWritable, Text, NullWritable, Text%3E %7B%0A%0A%09@Override%0A%09public void setup(Context context) throws IOException %7B%0A%0A%09%7D%0A%0A%09@Override%0A%09public void map(NullWritable key, Text value, Context context)%0A%09%09throws IOException %7B%0A%09    // parse the binary, convert it to JSON, and index that JSON%0A%09    try %7B%0A%09%09String id=%22123%22;%0A%09%09String bulkJsonFormat = %22%7B%5C%22index%5C%22:%7B %5C%22_index%5C%22:%5C%22tv_perf_v3%5C%22, %5C%22_type%5C%22:%5C%22voters%5C%22,%5C%22_id%5C%22:%5C%22%25s%5C%22 %7D %7D%5Cn%25s%5Cn%22;%0A%0A%09%09context.write(null, new Text(String.format(bulkJsonFormat, id, value)));%0A%0A%09    %7D catch (Exception e) %7B%0A%09%09throw new IOException(e);%0A%09    %7D%0A%09%7D%0A
+%09return (success ? 0 : 1);
 %0A   

add tools runner