<file_sep>package InvertedIndex;
import java.io.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.StringUtils;

public class InvertedIndexMapper extends Mapper<LongWritable,Text,Text,Text>{
	private Text docID;
	private Text word=new Text();
	
	protected void setup(Context context){
		String filename=((FileSplit)context.getInputSplit()).getPath().getName();
		docID=new Text(filename);
	}
	protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException{
		String[] temp=value.toString().split(" ");		
		for(String token:temp){
			word.set(token);
			context.write(word,docID);
		}
	}
}
<file_sep>package MRUnitTest;
import static org.junit.Assert.*;
import org.junit.Test;

public class JunitTestCountA {

	@Test
	public void test() {
		JunitTest test=new JunitTest();
		int output=test.countA("AuthorA");
		assertEquals(2,output);
	}

}
<file_sep>package Titanic;

import java.io.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;

//The Reducer will count all the values for each unique Reducer.
//The Keys have been sorted by the gender column. 
public class TitanicReducer extends Reducer<TitanicCustomKey, IntWritable, TitanicCustomKey, IntWritable> {	 
	public void reduce(TitanicCustomKey key, Iterable<IntWritable> values, Context context)throws IOException, InterruptedException { 
		int sum = 0; 
		for (IntWritable val : values) { 
			sum += val.get(); 
		} 
		context.write(key, new IntWritable(sum)); 
	} 
}<file_sep>package practice; 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 

public class WholeFileISDriver  {
	public static void main(String[] args) throws Exception {
		if (args.length != 2) {
			System.out.println("Two parameters are required for DriverNLineInputFormat- <input dir> <output dir>\n");
			System.exit(-1);
		}

		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "whole file IS");
		job.getConfiguration().set("fs.file.impl", "WinLocalFileSystem"); 
		job.setJarByClass(practice.WholeFileISDriver.class);			
		
		job.setMapperClass(practice.WholeFileMapper.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		job.setNumReduceTasks(0);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		job.setInputFormatClass(practice.WholeFileInputFormat.class);
		//if you dont use combinetext input format, then number of mappers is equal to number of input files
		// job.setInputFormatClass(CombineTextInputFormat.class);
				
		FileOutputFormat.setOutputPath(job,new Path(args[1]));

		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
 
<file_sep>package CustomKeyValue;

import java.io.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;

public class CustomDataTypeMapper extends Mapper<LongWritable,Text,CustomKey,IntWritable>{
	private static final IntWritable one=new IntWritable(1);
	private CustomKey cKey=new CustomKey();
	
	private IntWritable reqNo=new IntWritable();
	private Text URL=new Text();
	private Text reqDate=new Text();
	private Text timestamp=new Text();
	private Text IP=new Text();
	
	public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException{
		String str[]=value.toString().split("\t");
		reqNo.set(Integer.parseInt(str[0]));
		URL.set(str[1]);
		reqDate.set(str[2]);
		timestamp.set(str[3]);
		IP.set(str[4]);
		//System.out.println(URL +"\t"+ reqDate+ "\t"+ timestamp+ "\t"+IP+ "\t"+ reqNo);
		//pass values to setter methods 
		/*cKey.setRegNo(regNo);
		cKey.setURL(URL);
		cKey.setRegDate(regDate);
		cKey.setTime(time);
		cKey.setIP(IP); */
		
		cKey.set(URL, reqDate, timestamp,IP, reqNo);
		// when you send CKey only IP is sent. because from getter only IP is taken rest all are commented
		context.write(cKey, one);
	}
}
<file_sep>package ChainingMRjobs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser; 

//method2: using job control for mapper1-->reducer1-->mapper2->redcuer2 
//you can break this Driver into two and run manually one by one by giving job 1 output directory to input of job2 as given below. but, this is the simplest mehtod
//hadoop jar chain.jar Job1 shakespeare output
//hadoop jar chain.jar Job2 output output2

//however, very similar way you can to mapper1-->mapper2->mapper3--> reducer1-->mapper4-->reducer2 ... in any fashion

public class JobChainDriverMethod2  {
	
	private static final String OUTPUT_PATH = "intermediate_output";

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();		
		if (otherArgs.length != 2) {
			System.err.println("Enter valid number of arguments <Inputdirectory>  <Outputlocation>");
			System.exit(-1);
		}
		Job job = Job.getInstance(conf); 
		job.getConfiguration().set("fs.file.impl", "WinLocalFileSystem");
		job.getConfiguration().set("mapreduce.output.textoutputformat.separator", "\t"); 
		
		Configuration conf1 = new Configuration();
	    Configuration conf2 = new Configuration();

	    Job job1 = new Job(conf1, "WC");
	    job1.getConfiguration().set("fs.file.impl", "WinLocalFileSystem");
	    job1.setJarByClass(ChainingMRjobs.JobChainDriverMethod2 .class);	    
	   
	    job1.setMapperClass(ChainingMRjobs.Mapper1.class);
	    job1.setReducerClass(ChainingMRjobs.Reducer1.class);
	    
	    job1.setOutputKeyClass(Text.class);
	    job1.setOutputValueClass(IntWritable.class);
	    
	    job1.setInputFormatClass(TextInputFormat.class);
		job1.setOutputFormatClass(TextOutputFormat.class);
	    ControlledJob cJob1 = new ControlledJob(conf1);
	    cJob1.setJob(job1);
	    FileInputFormat.addInputPath(job1, new Path(args[0]));
	    FileOutputFormat.setOutputPath(job1, new Path(OUTPUT_PATH));

	    Job job2 = new Job(conf2, "WC of WC");
	    job2.getConfiguration().set("fs.file.impl", "WinLocalFileSystem");
	    job2.setJarByClass(ChainingMRjobs.JobChainDriverMethod2.class);	    
		
	    job2.setMapperClass(ChainingMRjobs.Mapper2.class);
	    job2.setReducerClass(ChainingMRjobs.Reducer2.class);
	    //job2.setSortComparatorClass(ReverseOrder.class); 
	    
	    job2.setOutputKeyClass(Text.class);
	    job2.setOutputValueClass(IntWritable.class);
	    
	    job2.setInputFormatClass(TextInputFormat.class);
		job2.setOutputFormatClass(TextOutputFormat.class);
		
		ControlledJob cJob2 = new ControlledJob(conf2);
	    cJob2.setJob(job2);
	    FileInputFormat.addInputPath(job2, new Path(OUTPUT_PATH+"/part*"));
	    FileOutputFormat.setOutputPath(job2, new Path(args[1]));

	    JobControl jobctrl = new JobControl("jobctrl");
	    jobctrl.addJob(cJob1);
	    jobctrl.addJob(cJob2);
	    cJob2.addDependingJob(cJob1);
	    jobctrl.run();	
	    
	    // delete jobctrl.run();
	  /*  Thread t = new Thread(jobctrl);
	    t.start();
	    String oldStatusJ1 = null;
	    String oldStatusJ2 = null;
	    while (!jobctrl.allFinished()) {
	      String status =cJob1.toString();
	      String status2 =cJob2.toString();
	      if (!status.equals(oldStatusJ1)) {
	        System.out.println(status);
	        oldStatusJ1 = status;
	      }
	      if (!status2.equals(oldStatusJ2)) {
	        System.out.println(status2);
	        oldStatusJ2 = status2;
	      }     
	     }
	    System.exit(0);*/
	}	
}
<file_sep>package DistriCache;
import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer; 

public class NcdcMaxTempReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
	
	private NcdcStationMetadataParcer metadata;
	private Configuration conf;
	String fileName ;
	String stationName;
	
	protected void setup(Context context) throws IOException{
		try{
			URI[] patternsURIs = Job.getInstance(conf).getCacheFiles(); 
			for (URI patternsURI : patternsURIs) { 
    			Path patternsPath = new Path(patternsURI.getPath());
    			fileName = patternsPath.getName().toString().trim(); 
    			if (fileName.equals("station.txt")){
					metadata = new NcdcStationMetadataParcer();
					metadata.initialize(patternsPath);
					break;
				}
			}			
			System.out.println("File : "+ patternsURIs[0].toString());
		}
		catch(NullPointerException e){
			System.out.println("Exception : "+e);
		}		
		System.out.println("Cache : "+context.getConfiguration().get("mapred.cache.files"));
	}
	
	protected void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException, InterruptedException{	

		System.out.println("step1");
		stationName = metadata.getStationName(key.toString());

		System.out.println("step2");
		int maxValue = Integer.MIN_VALUE;
		for (IntWritable val:values){
			maxValue = Math.max(maxValue, val.get());
		}
		context.write(new Text(stationName),new IntWritable(maxValue));
	}
}
<file_sep>package FileSystemAPI;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;

//dont execute this program from eclispe on windows as permission error will be the result
//execute this program in hadoop environment or eclipse on hadoop

public class FileDelete {

	public static void main(String[] args)throws IOException {
		// TODO Auto-generated method stub
		Configuration conf=new Configuration();
		String uri="hdfs://node2:10001/";
		Path path = new Path(uri); 
		FileSystem fs=FileSystem.get(URI.create(uri),conf);
		
		//display files and directories
		FileStatus[] status=fs.listStatus(path);  // to show list of files in a directory
		Path[] paths=FileUtil.stat2Paths(status);
		for(Path p:paths){
			System.out.println(p.toString()); 
		}	
		//delete input.txt
		Path path1 = new Path(uri+"/input.txt"); 
		path.getFileSystem(conf).delete(path1); 
		
		Path path2 = new Path(uri); 
		FileStatus[] status1=fs.listStatus(path2);  // to show list of files in a directory
		Path[] paths1=FileUtil.stat2Paths(status);
		//display files and directories
		for(Path p:paths1){
			System.out.println(p.toString()); 
		}	
	}

}
<file_sep>package CustomKeyValue;

import java.io.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;

// output of mapper is key:/vfnlyvnx.html  value: 172.16.31.10 and 1

public class MultiWritableReducer extends Reducer<Text,MultiWritableValue,Text,IntWritable>{
	
	private Text result = new Text();
	public void reduce(Text key,Iterable<MultiWritableValue> values, Context context) throws IOException, InterruptedException{
		
		int sum = 0;
		StringBuilder output = new StringBuilder();
		
		for (MultiWritableValue multi : values) {
			Writable writable = multi.get();
			if (writable instanceof IntWritable){
				sum += ((IntWritable)writable).get();
			}
			/*else{
				requests.append(((Text)writable).toString());
				requests.append("\t");
			}*/
		}		
		//result.set(key + "\t"+sum);
		context.write(key, new IntWritable(sum));
	}
}
<file_sep>package DistriCache;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
/*
  in NcdcMaxTempReducer it shows NullPointer Exception. understand the program and debug it. 
  input files are WeatherRecord.txt and station.txt
  add MRv2 jars. dont add MRv1 it will lead to exception.
  run in hadoop services. it doesnot run in Eclipse
 */

public class NCDCDriver{
		
	public static void main(String[] arg) throws Exception {
		if(arg.length !=2) {
			System.err.println("Usage:<inputpath> <outputpath>");
			System.exit(-1);
		}	

		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "Distributed Cache");
		job.setJarByClass(DistriCache.NCDCDriver.class);			
			
		job.setMapperClass(DistriCache.NcdcMaxTempMapper.class);
		//job.setCombinerClass(DistriCache.NcdcMaxTempCombiner.class);
		job.setReducerClass(DistriCache.NcdcMaxTempReducer.class);
			
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(arg[0]));
		FileOutputFormat.setOutputPath(job, new Path(arg[1])); 
		
		//for MRv1 with its libraries
		//DistributedCache.addCacheFile(new URI("hdfs://node1:10001/input/station.txt"), job.getConfiguration()); 
		
		//for MRv2 with its libraries
		job.addCacheFile(new Path("hdfs://node1:10001/input/station.txt").toUri());  
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
<file_sep>package RunTimeArguments;
import java.io.IOException; 
import org.apache.hadoop.io.*;  		  
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.Mapper.Context; 

public class WCMRv2Reducer extends Reducer<Text,IntWritable,Text,IntWritable> {
	private IntWritable result = new IntWritable();
	int lower;
	int upper;
	
	//set up method is called only once after JVM created
	public void setup(Context context)throws IOException, InterruptedException{
		lower = Integer.parseInt(context.getConfiguration().get("LowerLimit"));
		upper = Integer.parseInt(context.getConfiguration().get("UpperLimit"));
	}
	
	public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
		int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			if(sum>=lower && sum<=upper){
				result.set(sum);
				context.write(key, result);
			}			
	}
	
	// cleanup method is called only once before JVM exit
	public void cleanup(Context context)throws IOException,InterruptedException{
		
	}
}
<file_sep>package MRUnitTest;
import static org.junit.Assert.*;
import org.junit.Test;
//https://www.youtube.com/watch?v=I8XXfgF9GSc
public class JunitSquareTest {

	@Test
	public void test() {
		JunitTest test=new JunitTest();
		int output=test.square(5);
		assertEquals(26,output);
	}

}
<file_sep>package practice;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  

public class CombineFilesDriver {
	public static void main(String[] args) throws Exception { 
		Configuration conf = new Configuration(); 
		Job job = Job.getInstance(conf, "word count");
		//or Job job=Job.getInstance(getConf());
		job.getConfiguration().set("fs.file.impl", "WinLocalFileSystem");
		job.setJarByClass(practice.CombineFilesDriver.class);  		
		
		job.setMapperClass(practice.WCMRv2Mapper.class);	// check number of map output and it is not sorted
		//job.setReducerClass(practice.WCMRv2Reducer.class); 
		job.setNumReduceTasks(0);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class); 
		
		job.setInputFormatClass(CombineTextInputFormat.class);  // just enable this if you have more number of small files to pack as one IS.  It is in hadoop-mapreduce-client-core-2.7.0
		CombineTextInputFormat.setMaxInputSplitSize(job, 64000000);		// many files are put into 64 MB
		
		FileInputFormat.setInputPaths(job, new Path("D:/Eclipse Project/Practice/dataset/many files/*.txt"));
		FileOutputFormat.setOutputPath(job,new Path("D:/Eclipse Project/Practice/output"));
		
		System.exit( job.waitForCompletion(true) ? 0 : 1);
	}	
}

 <file_sep>package Titanic;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.InputSplit; 
import org.apache.hadoop.mapreduce.RecordReader; 
import org.apache.hadoop.mapreduce.TaskAttemptContext; 
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader; 

public class TitanicRR extends RecordReader<TitanicCustomKey,IntWritable> { 
	
	private TitanicCustomKey key; 
	private IntWritable value; 
	private LineRecordReader reader = new LineRecordReader(); 
			
	@Override 
	public TitanicCustomKey getCurrentKey() throws IOException, InterruptedException {
		// TODO Auto-generated method stub 
		return key; 
	} 
	
	@Override 
	public IntWritable getCurrentValue() throws IOException, InterruptedException { 
		// TODO Auto-generated method stub 
		return value; 
	} 
	
	@Override 
	public float getProgress() throws IOException, InterruptedException { 
		// TODO Auto-generated method stub 
		return reader.getProgress(); 
	} 
	
	@Override 
	public void initialize(InputSplit is, TaskAttemptContext tac)throws IOException, InterruptedException { 
		reader.initialize(is, tac); 
	}
	
	@Override 
	//every key:value from dataset is passed to mapper
	
	public boolean nextKeyValue() throws IOException, InterruptedException { 
		// TODO Auto-generated method stub 
		boolean gotNextKeyValue = reader.nextKeyValue(); 
		if(gotNextKeyValue){ 
			if(key==null){ 
				key = new TitanicCustomKey(); 
			} 
			if(value == null){ 
				value = new IntWritable(); 
			} 
			Text line = reader.getCurrentValue(); 
			String[] tokens = line.toString().split(","); 
			//As discussed earlier, we need 2nd and 5th columns passed to our custom key.
			//The value is set as ‘1‘ since we need to count the number of people. so need not assign value 1 in mapper
			key.setX(new String(tokens[1])); 			
			key.setY(new String(tokens[4])); 
			value.set(new Integer(1)); 
		} 
		else { 
			key = null; 
			value = null; 
		}
		return gotNextKeyValue; 
	} 
	@Override 
	public void close() throws IOException { 
		// TODO Auto-generated method stub 
		reader.close(); 
	} 
}
<file_sep>package cards;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer; 

public class CardRecordCount{
	
	private static class RecordMapper extends Mapper <LongWritable,Text,Text,IntWritable>{
		public void map(LongWritable lineoffset, Text record, Context output)throws IOException, InterruptedException{
			output.write(new Text("count"), new IntWritable(1));
		}
	}
	
	private static class RecordReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
		public void reduce(Text key,Iterable<IntWritable> values, Context output)throws IOException, InterruptedException{
			int sum = 0;
		      for (IntWritable val : values) {
		        sum += val.get();
		      }
		     output.write(new Text("total counte \t"), new IntWritable(sum));
		}
	}
	public static void main(String args[]) throws Exception{
		// TODO Auto-generated method stub
		if(args.length <2) {
			System.err.println("Usage:<inputpath> <outputpath>");
			System.exit(-1);
		}
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.getConfiguration().set("fs.file.impl", "WinLocalFileSystem");		
		
		job.setMapperClass(RecordMapper.class);
		job.setReducerClass(RecordReducer.class);  // user defined class for counting
		//job.setReducerClass(LongSumReducer.class);  //inbuilt class for counting.
		//job.setNumReduceTasks(0);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);		
	}

}<file_sep>package FileSystemAPI;

import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;

public class ListFiles {
	public static void main(String args[])throws Exception{
		//pass "src/cards/" as command line arguments. 
		String uri="hdfs://node2:10001/"; //args[0];
		Configuration conf=new Configuration();	
		   
		//convert path string from URI so that can manipulate with many inbuilt functions
		FileSystem fs=FileSystem.get(URI.create(uri),conf);
		Path path=new Path(uri);
		
		FileStatus[] status=fs.listStatus(path);  // to show list of files in a directory
		Path[] paths=FileUtil.stat2Paths(status);
		for(Path p:paths){
			System.out.println(p.toString()); 
		}	
		
	}
}
<file_sep>package cards;


import java.io.IOException;

import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;  
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.output.FileOutputFormat; 
import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer; 

public class CardCountBySuite {

	public static class Map extends Mapper<LongWritable,Text,Text,LongWritable>{
		public void map(LongWritable key, Text value, Context output) throws IOException, InterruptedException{
			output.write(new Text(value.toString().split("\\|")[1]), new LongWritable(1));
		}
	}
	
	public static void main(String args[]) throws Exception{
		if(args.length <2) {
			System.err.println("Usage:<inputpath> <outputpath>");
			System.exit(-1);
		}
		Configuration conf=new Configuration();
		Job job=Job.getInstance(conf); 
		job.getConfiguration().set("fs.file.impl", "WinLocalFileSystem");
		
		job.setMapperClass(Map.class);	
		job.setReducerClass(LongSumReducer.class);
		//job.setPartitionerClass(HashPartitioner.class);
		//job.setNumReduceTasks(0);
		
		//job.setMapOutputKeyClass(Text.class);
		//job.setMapOutputValueClass(LongWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job,new Path(args[1]));
		
		System.exit(job.waitForCompletion(true) ? 0 : 1); 
	}
}
<file_sep>package Filtering;

import java.io.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;  // it is in hadoop-mapreduce-client-core-2.7.0
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import NYSE.CustomKeyValue.*;
import NYSE.AvgStockVolPerMonth.*; 

public class AvgStockVolMonthDriver extends Configured implements Tool{
	 public int run(String args[]) throws IOException, InterruptedException, ClassNotFoundException {
		 if (args.length < 2) {
	    	System.err.println("Usage: wordcount <in> [<in>...] <out>");
	    	System.exit(2);
	    }    
		Configuration conf=getConf(); 
		Job job=Job.getInstance(conf,"NYSE average stock volume per month");
		//job.getConfiguration().set("fs.file.impl", "WinLocalFileSystem");
		//job.setJarByClass(getClass());
		job.setJarByClass(NYSE.AvgStockVolPerMonth.AvgStockVolMonthDriver.class);
		
		job.setMapperClass(AvgStockMapperWithSetUp.class);
		//average of average is not right. so we define new combiner class rather reusing reducer task
		job.setCombinerClass(AvgStockVolMonthCombiner.class);
		//now check combiner output		
		job.setReducerClass(AvgStockVolMonthReducer.class);
		job.setNumReduceTasks(4);  // to check mapper output
		
		job.setMapOutputKeyClass(TextPair.class);
		job.setMapOutputValueClass(LongPair.class);  		
		job.setOutputKeyClass(TextPair.class);
		job.setOutputValueClass(LongPair.class); 
		
		job.setInputFormatClass(CombineTextInputFormat.class);  // just enable this if you have more number of small files to pack as one IS.  It is in hadoop-mapreduce-client-core-2.7.0
		CombineTextInputFormat.setMaxInputSplitSize(job, 64000000);		// many files are put into 64 MB
		//job.setInputFormatClass(TextInputFormat.class); 
		//job.setOutputFormatClass(TextOutputFormat.class);
				
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		for (int i = 0; i < otherArgs.length - 1; ++i) {
		      FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
		}
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1])); 
	/*	FileInputFormat.addInputPath(job, new Path("D:/Eclipse Project/Practice/dataset/nyse_data/*.csv"));
	 	FileOutputFormat.setOutputPath(job, new Path("D:/Eclipse Project/Practice/output"));*/
						
		return job.waitForCompletion(true) ? 0 : 1;
	} 
	public static void main(String args[])throws Exception{ 
		int exitCode=ToolRunner.run(new AvgStockVolMonthDriver(), args);
		 System.exit(exitCode);	    
	}
}
