package com.thinkbiganalytics.esbulkloader;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
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.TextInputFormat;

public class ElasticSearchBulkLoader {
    public static void main(String args[]) throws Exception {
	Configuration conf = new Configuration();

	Job job = new Job(conf, "add_files_to_es");
	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);

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

    }

}

@@ -89,33 +89,152 @@
 che.
-hadoop.conf.Configuration
+commons.logging.Log;%0Aimport org.apache.commons.logging.LogFactory;%0Aimport org.apache.hadoop.conf.Configuration;%0Aimport org.apache.hadoop.fs.Path
 ;%0Aim
@@ -439,126 +439,405 @@
 put.
-TextInputFormat;%0A%0Apublic class ElasticSearchBulkLoader %7B%0A    public static void main(String args%5B%5D) throws Exception %7B
+FileInputFormat;%0Aimport org.apache.hadoop.mapreduce.lib.input.TextInputFormat;%0Aimport org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;%0A%0Apublic class ElasticSearchBulkLoader %7B%0A    %0A    private static Log log = LogFactory.getLog(ElasticSearchBulkLoader.class);%0A    public static void main(String args%5B%5D) throws Exception %7B%0A%09%0A%09log.info(String.format(%22%25s %25s %25s%22, args%5B0%5D, args%5B1%5D, args%5B2%5D));%0A%09
 %0A%09Co
@@ -906,25 +906,15 @@
 nf, 
-%22add_files_to_es%22
+args%5B0%5D
 );%0A%09
@@ -1229,16 +1229,132 @@
 .class);
+%0A%09%0A%09 FileInputFormat.addInputPath(job, new Path(args%5B1%5D));%0A%09 FileOutputFormat.setOutputPath(job, new Path(args%5B2%5D));
 %0A%0A%09// ac

Add input paths and arg processing