package nz.org.hikari.cat.ncdc_count;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

// Driver for NCDC record count MapReduce job
//
// Args: <input location> <output location> [number of reduce tasks]
//
//       e.g. s3://BUCKETNAME/ncdc/data/ /out 1
//
//       (note that /out will be created in HDFS, not the local filesystem,
//       and that it must not exist prior to running the job).
//
// Input: NCDC data files, optionally compressed in a MapReduce-compatible
//        format such as gzip or bzip2.
//
// Output: Tab-separated text file(s) containing the NCDC station ID (USAF and
//         WBAN identifiers concatenated, e.g. 029070-99999) and number of
//         input records for that station. There will be one output file per
//         reducer task, and the record count for a station will never be zero
//         because only station IDs seen in the input are counted.

public class NcdcRecordCount extends Configured implements Tool {

	public int run(String[] args) throws Exception {
		Job job = Job.getInstance(getConf(), "NCDC Record Count");
		job.setJarByClass(NcdcRecordCount.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		job.setMapperClass(NcdcRecordCountMapper.class);
		job.setCombinerClass(LongSumReducer.class);
		job.setReducerClass(LongSumReducer.class);
		if (args.length == 3)
			job.setNumReduceTasks(Integer.parseInt(args[2]));
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		
		return job.waitForCompletion(true) ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		System.exit(ToolRunner.run(new NcdcRecordCount(), args));
	}

}

@@ -587,33 +587,8 @@
 ion%3E
- %5Bnumber of reduce tasks%5D
 %0A//%0A
@@ -632,18 +632,16 @@
 ta/ /out
- 1
 %0A//%0A//  
@@ -1500,128 +1500,8 @@
 s);%0A
-%09%09%0A%09%09FileInputFormat.addInputPath(job, new Path(args%5B0%5D));%0A%09%09FileOutputFormat.setOutputPath(job, new Path(args%5B1%5D));%0A%09%09%0A
 %09%09jo
@@ -1539,32 +1539,32 @@
 tMapper.class);%0A
+
 %09%09job.setCombine
@@ -1642,85 +1642,8 @@
 s);%0A
-%09%09if (args.length == 3)%0A%09%09%09job.setNumReduceTasks(Integer.parseInt(args%5B2%5D));%0A
 %09%09jo
@@ -1722,18 +1722,342 @@
 class);%0A
+%0A
 %09%09
+if (args.length %3C 2)%0A%09%09%09throw new IllegalArgumentException(%22Insufficient command-line arguments%22);%0A%09%09FileInputFormat.setInputDirRecursive(job, true);%0A%09%09for (int i = 0; i %3C args.length - 1; i++)%0A%09%09%09FileInputFormat.addInputPath(job, new Path(args%5Bi%5D));%0A%09%09FileOutputFormat.setOutputPath(job, new Path(args%5Bargs.length - 1%5D));%0A
 %0A%09%09retur

Allow multiple input paths