code: package edu.jhu.thrax.hadoop.tools;
  
  import java.util.Map;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  
  import edu.jhu.thrax.hadoop.datatypes.RuleWritable;
  import edu.jhu.thrax.hadoop.features.mapred.MapReduceFeature;
  import edu.jhu.thrax.hadoop.features.mapred.MapReduceFeatureFactory;
  import edu.jhu.thrax.util.ConfFileParser;
  
  public class FeatureTool extends Configured implements Tool
  {
      public int run(String [] argv) throws Exception
      {
          if (argv.length < 2) {
              System.err.println("usage: FeatureTool <conf file> <feature>");
              return 1;
          }
          String confFile = argv[0];
          String featureName = argv[1];
          MapReduceFeature f = MapReduceFeatureFactory.get(featureName);
          if (!(f instanceof MapReduceFeature)) {
              System.err.println("Not a MapReduceFeature: " + featureName);
              return 1;
          }
          Configuration conf = getConf();
          Map<String,String> options = ConfFileParser.parse(confFile);
          for (String opt : options.keySet()) {
              conf.set("thrax." + opt, options.get(opt));
          }
          String workDir = conf.get("thrax.work-dir");
          if (workDir == null) {
              System.err.println("set work-dir key in conf file " + confFile + "!");
              return 1;
          }
          if (!workDir.endsWith(Path.SEPARATOR)) {
              workDir += Path.SEPARATOR;
              conf.set("thrax.work-dir", workDir);
          }
          Job job = new Job(conf, String.format("thrax-%s", featureName));
  
          job.setJarByClass(f.getClass());
          job.setMapperClass(f.mapperClass());
          job.setCombinerClass(f.combinerClass());
          job.setSortComparatorClass(f.sortComparatorClass());
          job.setPartitionerClass(f.partitionerClass());
          job.setReducerClass(f.reducerClass());
  
          job.setInputFormatClass(SequenceFileInputFormat.class);
  
          job.setMapOutputKeyClass(RuleWritable.class);
          job.setMapOutputValueClass(IntWritable.class);
  
          job.setOutputKeyClass(RuleWritable.class);
          job.setOutputValueClass(IntWritable.class);
  
          job.setOutputFormatClass(SequenceFileOutputFormat.class);
  
          FileInputFormat.setInputPaths(job, new Path(workDir + "rules"));
          FileOutputFormat.setOutputPath(job, new Path(workDir + featureName));
  
          job.submit();
          return 0;
      }
  
      public static void main(String [] argv) throws Exception
      {
          int exit_code = ToolRunner.run(null, new FeatureTool(), argv);
          System.exit(exit_code);
      }
  }
  