Pasted

package tdd.tp03;
 
import java.io.IOException;
import java.util.Collection;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
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;
 
class SubstringCountMapper
extends Mapper<Object, Text, Text, NullWritable> {
 
    @Override
    public void map(Object o, Text in, Context ctx) 
            throws IOException, InterruptedException
    {
        for (String word : in.toString().toLowerCase().split("[ \n\t.,;:!?]+")) {
            ctx.write(new Text(word), NullWritable.get());
        }       
    }
    
}
 
class SubstringCountReducer
extends Reducer<Text, NullWritable, Text, IntWritable> {
 
    @Override
    public void reduce(Text str, Iterable<NullWritable> nulls,
                        Context ctx) 
    throws IOException, InterruptedException {
        
        if (nulls instanceof Collection) {
            ctx.write(str, new IntWritable(((Collection<?>)nulls).size()));
        } else {
            int total = 0;
            for (Object o : nulls) {
                total++;
            }
            ctx.write(str, new IntWritable(total));
        }
    }
    
}
    
public class DriverWordCount {
 
    
    
    public static void main(String[] args)
    
    {
        if (args.length < 2) System.exit(2);
        
        try {
            
            //Création d'une nouvelle configuration
            Configuration conf = new Configuration();
 
            Job job = Job.getInstance(conf, "tdd.tp03");
            
                    
            
            //Le nom de la classe 'main'
            job.setJarByClass(DriverWordCount.class);
 
            //Le format d'entrée du Map
            //TextInputFormat est un alias pour
            //InputFormat<LongWritable, Text>;
            
            job.setInputFormatClass(TextInputFormat.class);
            
            //Initialisation du Map
            job.setMapperClass(SubstringCountMapper.class);
            
            //Initialisation du Reduce
            job.setReducerClass(SubstringCountReducer.class);
 
            //Sortie du Map
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);
            
            //Sortie (du Reduce)
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
                        
            FileOutputFormat.setOutputPath(job, new Path(args[0]));
            
            for(int i = 1; i < args.length; i++)
                FileInputFormat.addInputPath(job, new Path(args[i]));
 
            
                        
            
            System.exit(job.waitForCompletion(true) ?  0 : 1);
            
        } catch (Exception e) {
            System.err.println("Erreur : " + e);
        }
        
    }
    
}