TutorialOneCompleteSourceListing

package org.apache.nutch.examples

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

import org.apache.nutch.parse.*;
import org.apache.nutch.util.*;

public class LinkCounter {

	public static class CounterMapper extends MapReduceBase implements Mapper
	{
		public void map(WritableComparable key, Writable value, OutputCollector collector, Reporter reporter) throws IOException {
			// TODO Auto-generated method stub
			ParseData data = (ParseData)value;
			
			IntWritable outboundLinkCount = new IntWritable(data.getOutlinks().length);					
			
			collector.collect(key, outboundLinkCount);
		}

		public void close() throws IOException {
			// TODO Auto-generated method stub
			super.close();
		}

		public void configure(JobConf arg0) {
			// TODO Auto-generated method stub
			super.configure(arg0);
		}
		
	}
	
	public static class CounterReducer extends MapReduceBase implements Reducer
	{

		public void reduce(WritableComparable url, Iterator iterator, OutputCollector output, Reporter reporter) throws IOException {
			IntWritable linkCount = (IntWritable)iterator.next();
			output.collect(url, linkCount);
		}

		public void close() throws IOException {
			// TODO Auto-generated method stub
			super.close();
		}

		public void configure(JobConf arg0) {
			// TODO Auto-generated method stub
			super.configure(arg0);
		}
		
	
	}
	
	public static void main(String[] args) throws IOException{
		Configuration config = NutchConfiguration.create();
		
	    JobConf jobConfig = new NutchJob(config);
	    jobConfig.setJobName("countlinks");
	 
	    jobConfig.setInputFormat(SequenceFileInputFormat.class);
	    
	    jobConfig.setOutputFormat(MapFileOutputFormat.class);
	    
	    // the keys are words (strings)
	    jobConfig.setOutputKeyClass(Text.class);
	    // the values are counts (ints)
	    jobConfig.setOutputValueClass(IntWritable.class);
	    
	    jobConfig.setMapperClass(CounterMapper.class);        
	    jobConfig.setCombinerClass(CounterReducer.class);
	    jobConfig.setReducerClass(CounterReducer.class);
	    
	    jobConfig.setInputPath(new Path((String) args[0], ParseData.DIR_NAME));
	    jobConfig.setOutputPath(new Path((String) args[1]));
	    
	    JobClient.runJob(jobConfig);
	}

}