code: package HiBench;
  
  import java.io.IOException;
  import java.net.URISyntaxException;
  import java.util.Random;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapred.FileInputFormat;
  import org.apache.hadoop.mapred.FileOutputFormat;
  import org.apache.hadoop.mapred.JobClient;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.MapReduceBase;
  import org.apache.hadoop.mapred.Mapper;
  import org.apache.hadoop.mapred.OutputCollector;
  import org.apache.hadoop.mapred.Reporter;
  import org.apache.hadoop.mapred.SequenceFileOutputFormat;
  import org.apache.hadoop.mapred.lib.NLineInputFormat;
  
  public class BayesData {
  
  	private static final Log log = LogFactory.getLog(BayesData.class.getName());
  	
  	private DataOptions options;
  	private Dummy dummy;
  	private int cgroups;
  	
  	BayesData(DataOptions options) {
  		this.options = options;
  		parseArgs(options.getRemainArgs());
  	}
  	
  	private void parseArgs(String[] args) {
  		
  		for (int i=0; i<args.length; i++) {
  			if ("-class".equals(args[i])) {
  				cgroups = Integer.parseInt(args[++i]);
  			} else {
  				DataOptions.printUsage("Unknown bayes data arguments -- " + args[i] + "!!!");
  				System.exit(-1);
  			}
  		}
  	}
  
  	private static class CreateBayesPages extends MapReduceBase implements
  	Mapper<LongWritable, Text, Text, Text> {
  
  		private static final Log log = LogFactory.getLog(CreateBayesPages.class.getName());
  		
  		private long pages, slotpages;
  		private int groups;
  		private HtmlCore generator;
  		private Random rand;
  
  		public void configure(JobConf job) {
  			try {
  				pages = job.getLong("pages", 0);
  				slotpages = job.getLong("slotpages", 0);
  				groups = job.getInt("groups", 0);
  				
  				generator = new HtmlCore(job);
  			} catch (IOException e) {
  				// TODO Auto-generated catch block
  				e.printStackTrace();
  			}
  		}
  
  		@Override
  		public void map(LongWritable key, Text value,
  				OutputCollector<Text, Text> output, Reporter reporter)
  				throws IOException {
  
  			int slotId = Integer.parseInt(value.toString().trim());
  			long[] range = HtmlCore.getPageRange(slotId, pages, slotpages);
  			generator.fireRandom(slotId);
  			rand = new Random(slotId * 1000 + 101);
  			
  			Text k = new Text();
  			for (long i=range[0]; i<range[1]; i++) {
  				String classname = "/class" + rand.nextInt(groups);
  				k.set(classname);
  				value.set(generator.genBayesWords());
  				output.collect(k, value);
  				reporter.incrCounter(HiBench.Counters.BYTES_DATA_GENERATED,
  					k.getLength()+value.getLength());
  				if (0==(i % 10000)) {
  					log.info("still running: " + (i - range[0]) + " of " + slotpages);
  				}
  			}
  		}
  	}
  	
  	private void setBayesOptions(JobConf job) throws URISyntaxException {
  		job.setLong("pages", options.getNumPages());
  		job.setLong("slotpages", options.getNumSlotPages());
  		job.setInt("groups", cgroups);
  		
  		Utils.shareWordZipfCore(options, job);
  	}
  	
  	private void createBayesData() throws IOException, URISyntaxException {
  		
  		log.info("creating bayes text data ... ");
  
  		JobConf job = new JobConf();
  
  		Path fout = options.getResultPath();
  		Utils.checkHdfsPath(fout);
  		
  		String jobname = "Create bayes data";
  		job.setJobName(jobname);
  
  		Utils.shareDict(options, job);
  		
  		setBayesOptions(job);
  		
  		FileInputFormat.setInputPaths(job, dummy.getPath());
  		job.setInputFormat(NLineInputFormat.class);
  
  		job.setJarByClass(CreateBayesPages.class);
  		job.setMapperClass(CreateBayesPages.class);
  		job.setNumReduceTasks(0);
  		
  		FileOutputFormat.setOutputPath(job, fout);
  		job.setOutputFormat(SequenceFileOutputFormat.class);
  		job.setMapOutputKeyClass(Text.class);
  		job.setMapOutputValueClass(Text.class);
  		job.setOutputKeyClass(Text.class);
  		job.setOutputValueClass(Text.class);
  
  		log.info("Running Job: " +jobname);
  		log.info("Pages file " + dummy.getPath() + " as input");
  		log.info("Rankings file " + fout + " as output");
  		JobClient.runJob(job);
  		log.info("Finished Running Job: " + jobname);
  	}
  	
  	private void init() throws IOException {
  
  		Utils.checkHdfsPath(options.getResultPath(), true);
  		Utils.checkHdfsPath(options.getWorkPath(), true);
  
  		dummy = new Dummy(options.getWorkPath(), options.getNumMaps());
  
  		int words = RawData.putDictToHdfs(new Path(options.getWorkPath(), HtmlCore.getDictName()), options.getNumWords());
  		options.setNumWords(words);
  		
  		Utils.serialWordZipf(options);
  	}
  
  	public void generate() throws Exception {
  		
  		init();
  		
  		createBayesData();
  		
  		close();
  	}
  
  	private void close() throws IOException {
  		log.info("Closing bayes data generator...");
  		Utils.checkHdfsPath(options.getWorkPath());
  	}
  }
  