code: package HiBench;
  
  import java.io.IOException;
  import java.net.URISyntaxException;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.Random;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.filecache.DistributedCache;
  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.MapFileOutputFormat;
  import org.apache.hadoop.mapred.MapReduceBase;
  import org.apache.hadoop.mapred.Mapper;
  import org.apache.hadoop.mapred.OutputCollector;
  import org.apache.hadoop.mapred.Reducer;
  import org.apache.hadoop.mapred.Reporter;
  import org.apache.hadoop.mapred.SequenceFileInputFormat;
  import org.apache.hadoop.mapred.SequenceFileOutputFormat;
  import org.apache.hadoop.mapred.TextInputFormat;
  import org.apache.hadoop.mapred.TextOutputFormat;
  import org.apache.hadoop.mapred.lib.MultipleInputs;
  import org.apache.hadoop.mapred.lib.NLineInputFormat;
  
  public class HiveData {
  
  	private static final Log log = LogFactory.getLog(HiveData.class.getName());
  	
  	private static final String RANKINGS = "rankings";
  	private static final String USERVISITS = "uservisits";
  	public static final String uagentf = "user_agents";
  	public static final String countryf = "country_codes";
  	public static final String searchkeyf = "search_keys";
  	
  	private DataOptions options;
  	private long visits;
  	
  	// client side delim
  	private String cdelim = ",";
  	private int chashsize = 150 * 1024 * 1024;
  	
  	private Dummy dummy;
  
  	HiveData(DataOptions options) {
  		this.options = options;
  		parseArgs(options.getRemainArgs());
  	}
  	
  	private void parseArgs(String[] args) {
  		
  		for (int i=0; i<args.length; i++) {
  			if ("-v".equals(args[i])) {
  				visits = Long.parseLong(args[++i]);
  			} else if ("-d".equals(args[i])) {
  				cdelim = args[++i];
  			} else {
  				DataOptions.printUsage("Unknown hive data arguments -- " + args[i] + "!!!");
  			}
  		}
  		
  		if (chashsize > options.getNumPages()) {
  			chashsize = (int) options.getNumPages();
  		}
  
  	}
  	
  	private void setRankingsOptions(JobConf job) throws URISyntaxException {
  		job.setLong("pages", options.getNumPages());
  		job.setLong("slotpages", options.getNumSlotPages());
  		job.set("delimiter", cdelim);
  		job.setInt("hashsize", chashsize);
  		Utils.shareLinkZipfCore(options, job);
  	}
  	
  	private void setVisitsOptions(JobConf job) {
  		job.setInt("slots", options.getNumMaps());
  		job.setLong("pages", options.getNumPages());
  		job.setLong("visits", visits);
  		job.set("delimiter", cdelim);
  	}
  
  	public static class DummyToRankingsMapper extends MapReduceBase implements
  	Mapper<LongWritable, Text, LongWritable, JoinBytesInt> {
  
  		private static final Log log = LogFactory.getLog(DummyToRankingsMapper.class.getName());
  		
  		private HtmlCore generator;
  		private long pages, slotpages;
  		private boolean outset;
  		private OutputCollector<LongWritable, JoinBytesInt> myout;
  		private JoinBytesInt uitem, ritem;
  		private short[] hash;
  		private HashMap<Integer, Integer> hm;
  		private int hashsize;
  		
  		private void getOptions(JobConf job) {
  			pages = job.getLong("pages", 0);
  			slotpages = job.getLong("slotpages", 0);
  			hashsize = job.getInt("hashsize", 0);
  		}
  
  		public void configure(JobConf job) {
  
  			getOptions(job);
  	
  			try {
  				generator = new HtmlCore(job);
  			} catch (IOException e) {
  				// TODO Auto-generated catch block
  				e.printStackTrace();
  			}
  
  			outset = false;
  			myout = null;
  			uitem = new JoinBytesInt();
  			uitem.url = new byte[HtmlCore.getMaxUrlLength()];
  			ritem = new JoinBytesInt();
  			ritem.refs = 1;
  			
  			hash = new short[hashsize];
  			hm = new HashMap<Integer, Integer>();
  		}
  	
  		public void map(LongWritable key, Text value, OutputCollector<LongWritable, JoinBytesInt> output,
  				Reporter reporter) throws IOException {
  
  			if (!outset) {
  				myout = output;
  				outset = true;
  			}
  
  			int slotId = Integer.parseInt(value.toString().trim());
  			generator.fireRandom(slotId);
  
  			long[] range = HtmlCore.getPageRange(slotId, pages, slotpages);
  
  			/**
  			 * For output collect
  			 */
  			for (long i=range[0]; i<range[1]; i++) {
  				key.set(i);
  
  				generator.nextUrlJoinBytesInt(uitem);
  				output.collect(key, uitem);
  				
  				long[] linkids = generator.genPureLinkIds();
  				for (int j=0; j<linkids.length; j++) {
  					long uid = linkids[j];
  					if (uid < hashsize) {
  						int iid = (int) uid;
  						if (hash[iid]>=0) {
  							if (hash[iid]==HtmlCore.MAX_SHORT) {
  								hm.put(iid, (int) (hash[iid]) + 1);
  								hash[iid] = -1;
  							} else {
  								hash[iid]++;
  							}
  						} else {
  							hm.put(iid, hm.get(iid) + 1);
  						}
  					} else {
  						key.set(uid); 
  						output.collect(key, ritem);
  					}
  				}
  				
  				if (0==(i % 10000)) {
  					log.info("still running: " + (i - range[0]) + " of " + slotpages);
  				}
  			}
  		}
  		
  		@Override
  		public void close ()
  		{
  			try {
  				LongWritable k = new LongWritable();
  				for (int i=0; i<hash.length; i++) {
  					if (hash[i] > 0) {
  						k.set(i);
  						ritem.refs = hash[i];
  						myout.collect(k, ritem);
  					} else if (hash[i] < 0) {
  						k.set(i);
  						ritem.refs = hm.get(i);
  						myout.collect(k, ritem);
  					}
  				}
  			} catch (IOException e) {
  				// TODO Auto-generated catch block
  				e.printStackTrace();
  			}
  		}
  	}
  
  	public static class JoinBytesIntCombiner extends MapReduceBase implements
  	Reducer<LongWritable, JoinBytesInt, LongWritable, JoinBytesInt> {
  
  //		Log log = null;
  		JoinBytesInt item;
  		
  		@Override
  		public void configure (JobConf job)
  		{
  			item = new JoinBytesInt();
  //			log = LogFactory.getLog(JoinBytesIntCombiner.class.getName());
  		}
  
  		@Override
  		public void reduce(LongWritable key, Iterator<JoinBytesInt> values,
  				OutputCollector<LongWritable, JoinBytesInt> output, Reporter reporter) throws IOException {
  
  			item.clear();
  //			StringBuffer sb =  new StringBuffer("Combine: " + v.toString());
  			while (values.hasNext()) {
  				item.add(values.next());
  //				sb.append("-> " + v.toString());
  			}
  			output.collect(key, item);
  //			log.info(sb);
  		}
  	}
  	
  	public static class GenerateRankingsReducer extends MapReduceBase implements
  	Reducer<LongWritable, JoinBytesInt, LongWritable, Text> {
  		
  		private static final Log log = LogFactory.getLog(GenerateRankingsReducer.class.getName());
  
  		private Random rand;
  		private int errors, missed;
  		private JoinBytesInt v;
  		private int pid;
  		
  		// job side delimiter
  		private String delim;
  //		private String missedids;
  		
  		public void configure (JobConf job)
  		{
  			delim = job.get("delimiter");
  			pid = job.getInt("mapred.task.partition", 0);
  			rand = new Random(pid + 1);
  
  			v = new JoinBytesInt();
  
  			errors = 0;
  			missed = 0;
  //			missedids = "";
  		}
  
  		public void close ()
  		{
  			log.info("pid: " + pid + ", " + errors + " erros, " + missed + " missed");
  		}
  
  		@Override
  		public void reduce(LongWritable key, Iterator<JoinBytesInt> values,
  				OutputCollector<LongWritable, Text> output, Reporter reporter) throws IOException {
  
  			v.clear();
  			while (values.hasNext()) {
  				v.add(values.next());
  			}
  			
  			if (0!=v.ulen) {
  				if (v.refs > 0) {
  					Text value = new Text(
  							new String(v.url) +
  							delim +
  							v.refs +
  							delim +
  							(rand.nextInt(99) + 1)
  							);
  					output.collect(
  							key, value);
  							
  					reporter.incrCounter(HiBench.Counters.BYTES_DATA_GENERATED, 8+value.getLength());
  				} else {
  					missed++;
  				}
  			} else {
  				errors++;					
  			}
  		}
  	}
  
  	private void createRankingsTableDirectly() throws IOException, URISyntaxException {
  
  		log.info("Creating table rankings...");
  
  		Path fout = new Path(options.getResultPath(), RANKINGS);
  
  		JobConf job = new JobConf(HiveData.class);
  		String jobname = "Create rankings";
  
  		/** TODO: change another more effective way as this operation may cause
  		 *  about 2 min delay (originally ~15min in total)
  		 */
  		setRankingsOptions(job);
  		job.setJobName(jobname);
  		job.set("mapred.reduce.slowstart.completed.maps", "0.3");
  		job.set("mapreduce.job.reduce.slowstart.completedmaps", "0.3");
  
  		job.setOutputKeyClass(LongWritable.class);
  		job.setOutputValueClass(Text.class);
  
  		job.setMapOutputKeyClass(LongWritable.class);
  		job.setMapOutputValueClass(JoinBytesInt.class);
  
  		job.setJarByClass(DummyToRankingsMapper.class);
  		job.setJarByClass(JoinBytesIntCombiner.class);
  		job.setJarByClass(GenerateRankingsReducer.class);
  		
  		job.setMapperClass(DummyToRankingsMapper.class);
  		job.setCombinerClass(JoinBytesIntCombiner.class);
  		job.setReducerClass(GenerateRankingsReducer.class);
  
  		if (options.getNumReds() > 0) {
  			job.setNumReduceTasks(options.getNumReds());
  		} else {
  			job.setNumReduceTasks(Utils.getMaxNumReds());
  		}
  
  		job.setInputFormat(NLineInputFormat.class);
  		FileInputFormat.setInputPaths(job, dummy.getPath());
  
  		 job.set("mapred.map.output.compression.type", "BLOCK");
  	 	 job.set("mapreduce.output.fileoutputformat.compress.type","BLOCK");
  		 MapFileOutputFormat.setCompressOutput(job, true);
  //		MapFileOutputFormat.setOutputCompressorClass(job, org.apache.hadoop.io.compress.LzoCodec.class);
  		 MapFileOutputFormat.setOutputCompressorClass(job, org.apache.hadoop.io.compress.DefaultCodec.class);
  
  		if (options.isSequenceOut()) {
  			job.setOutputFormat(SequenceFileOutputFormat.class);
  		} else {
  			job.setOutputFormat(TextOutputFormat.class);
  		}
  		
  		if (null != options.getCodecClass()) {
  			job.set("mapred.output.compression.type","BLOCK");
  		 	job.set("mapreduce.output.fileoutputformat.compress.type","BLOCK");	
  			FileOutputFormat.setCompressOutput(job, true);
  			FileOutputFormat.setOutputCompressorClass(job, options.getCodecClass());
  		}
  
  		FileOutputFormat.setOutputPath(job, fout);
  
  		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);
  	}
  
  	/***
  	 * Mapper to randomly create user visits. In map step, only the target
  	 * urls of user visits are created, the rest content of visits will be
  	 * created in reduce step
  	 * @author lyi2
  	 *
  	 */
  	public static class DummyToAccessNoMapper extends MapReduceBase implements
  	Mapper<LongWritable, Text, LongWritable, JoinBytesInt> {
  	
  		private JoinBytesInt vitem;
  		private long pages;
  		private long slots;
  		private long visits;
  		
  		// job side delimiter
  		private String delim;
  		private Visit visit;
  	
  		public void configure (JobConf job)
  		{
  			try {
  				pages = job.getLong("pages", 0);
  				slots = job.getLong("slots", 0);
  				visits = job.getLong("visits", 0);
  				delim = job.get("delimiter");
  	
  				visit = new Visit(DistributedCache.getLocalCacheFiles(job),
  						delim, pages);
  				
  				vitem = new JoinBytesInt();
  				vitem.refs = 1;
  	
  			} catch (IOException e) {
  				e.printStackTrace();
  			}
  		}
  	
  		@Override
  		public void map(LongWritable key, Text value,
  				OutputCollector<LongWritable, JoinBytesInt> output, Reporter reporter)
  						throws IOException {
  	
  			int slotId = Integer.parseInt(value.toString().trim());
  			visit.fireRandom(slotId);
  
  			for (long i=slotId; i<=visits;) {
  				// simply setting url id is fine in map step
  				key.set(visit.nextUrlId());
  				output.collect(key, vitem);
  				i = i + slots;
  			}
  		}
  	}
  
  	public static class SequenceRankingsToUrlsMapper extends MapReduceBase implements
  	Mapper<LongWritable, Text, LongWritable, JoinBytesInt> {
  		public JoinBytesInt uitem;
  		
  		public void configure(JobConf job) {
  			uitem = new JoinBytesInt();
  //			getBasicOptions(job);
  		}
  	
  		@Override
  		public void map(LongWritable key, Text value,
  				OutputCollector<LongWritable, JoinBytesInt> output, Reporter reporter) throws IOException {
  			
  			uitem.url= value.toString().split(",")[0].getBytes();
  			uitem.ulen = (byte) uitem.url.length;
  			
  			output.collect(key, uitem);
  		}
  	}
  	
  	public static class TextRankingsToUrlsMapper extends MapReduceBase implements
  	Mapper<LongWritable, Text, LongWritable, JoinBytesInt> {
  		public JoinBytesInt uitem;
  		
  		public void configure(JobConf job) {
  			uitem = new JoinBytesInt();
  //			getBasicOptions(job);
  		}
  	
  		@Override
  		public void map(LongWritable key, Text value,
  				OutputCollector<LongWritable, JoinBytesInt> output, Reporter reporter) throws IOException {
  			
  			String[] items = value.toString().split("[,\t]");
  			key.set(Long.parseLong(items[0]));
  			uitem.url= items[1].getBytes();
  			uitem.ulen = (byte) uitem.url.length;
  			
  			output.collect(key, uitem);
  		}
  	}
  	
  	public static class CreateUserVisitsReducer extends MapReduceBase implements
  	Reducer<LongWritable, JoinBytesInt, LongWritable, Text> {
  	
  		private static final Log log = LogFactory.getLog(CreateUserVisitsReducer.class.getName());
  		
  		private long pages;
  		private Visit visit;
  
  		private int errors, missed;
  		private JoinBytesInt vitem;
  		
  		// job side delimiter
  		private String delim;
  		private int pid;
  	
  		public void configure (JobConf job)
  		{
  			try {
  				pages = job.getLong("pages", 0);
  				delim = job.get("delimiter");
  				pid = job.getInt("mapred.task.partition", 0);
  	
  				visit = new Visit(DistributedCache.getLocalCacheFiles(job),
  						delim, pages);
  				visit.fireRandom(pid + 1);
  				
  				vitem = new JoinBytesInt();
  				
  				errors = 0;
  				missed = 0;
  	
  			} catch (IOException e) {
  				e.printStackTrace();
  			}
  		}
  	
  
  		public void close ()
  		{
  			log.info("pid: " + pid + ", " + errors + " erros, " + missed + " missed");
  		}
  
  		/**
  		 * Reduce: to sum up the record sizes (of slots) one by one so that to determine the
  		 * corresponding start point to hold the records for each slot.
  		 */
  		@Override
  		public void reduce(LongWritable key, Iterator<JoinBytesInt> values,
  				OutputCollector<LongWritable, Text> output, Reporter reporter) throws IOException {
  			
  			vitem.clear();
  //			StringBuffer sb = new StringBuffer("Reduce: " + v.toString());
  			while (values.hasNext()) {
  				vitem.add(values.next());
  //				sb.append("-> " + v.toString());
  			}
  //			log.info(sb);
  			
  			if (0!=vitem.ulen) {
  				if (vitem.refs > 0) {
  					for (int i=0; i<vitem.refs; i++) {
  						Text value = new Text(visit.nextAccess(new String(vitem.url)));
  						output.collect(key, value);
  						reporter.incrCounter(HiBench.Counters.BYTES_DATA_GENERATED, 8+value.getLength());
  					}
  				} else {
  					missed++;
  				}
  			} else {
  				errors++;					
  			}
  		}
  	}
  
  	private void createUserVisitsTableDirectly() throws IOException, URISyntaxException {
  
  		log.info("Creating user visits...");
  
  		Path rankings = new Path(options.getResultPath(), RANKINGS);
  		Path fout = new Path(options.getResultPath(), USERVISITS);
  
  		JobConf job = new JobConf(HiveData.class);
  		String jobname = "Create uservisits";
  		job.setJobName(jobname);
  		setVisitsOptions(job);
  		
  		/***
  		 * Set distributed cache file for table generation,
  		 * cache files include:
  		 * 1. user agents
  		 * 2. country code and language code
  		 * 3. search keys
  		 */
  
  		Path uagentPath = new Path(options.getWorkPath(), uagentf);
  		DistributedCache.addCacheFile(uagentPath.toUri(), job);
  		
  		Path countryPath = new Path(options.getWorkPath(), countryf);
  		DistributedCache.addCacheFile(countryPath.toUri(), job);
  
  		Path searchkeyPath = new Path(options.getWorkPath(), searchkeyf);
  		DistributedCache.addCacheFile(searchkeyPath.toUri(), job);
  
  		job.setOutputKeyClass(LongWritable.class);
  		job.setOutputValueClass(Text.class);
  
  		job.setMapOutputKeyClass(LongWritable.class);
  		job.setMapOutputValueClass(JoinBytesInt.class);
  
  		MultipleInputs.addInputPath(job, dummy.getPath(),
  				NLineInputFormat.class, DummyToAccessNoMapper.class);
  
  		if (options.isSequenceOut()) {
  			MultipleInputs.addInputPath(job, rankings,
  					SequenceFileInputFormat.class, SequenceRankingsToUrlsMapper.class);
  		} else {
  			MultipleInputs.addInputPath(job, rankings,
  					TextInputFormat.class, TextRankingsToUrlsMapper.class);
  		}
  
  		job.setCombinerClass(JoinBytesIntCombiner.class);
  		job.setReducerClass(CreateUserVisitsReducer.class);
  		
  		if (options.getNumReds() > 0) {
  			job.setNumReduceTasks(options.getNumReds());
  		} else {
  			job.setNumReduceTasks(Utils.getMaxNumReds());
  		}
  
  //		job.setNumReduceTasks(options.slots/2);
  
  		if (options.isSequenceOut()) {
  			job.setOutputFormat(SequenceFileOutputFormat.class);
  		} else {
  			job.setOutputFormat(TextOutputFormat.class);
  		}
  		
  		if (null != options.getCodecClass()) {
  			job.set("mapred.output.compression.type","BLOCK");
                          job.set("mapreduce.output.fileoutputformat.compress.type","BLOCK");
  			FileOutputFormat.setCompressOutput(job, true);
  			FileOutputFormat.setOutputCompressorClass(job, options.getCodecClass());
  		}
  		
  		FileOutputFormat.setOutputPath(job, fout);
  		
  		log.info("Running Job: " +jobname);
  		log.info("Dummy file " + dummy.getPath() + " as input");
  		log.info("Rankings file " + rankings + " as input");
  		log.info("Ouput file " + fout);
  		JobClient.runJob(job);
  		log.info("Finished Running Job: " + jobname);
  	}
  
  	public void generate() throws Exception {
  		
  		log.info("Generating hive data files...");
  		init();
  		
  		createRankingsTableDirectly();
  		createUserVisitsTableDirectly();
  		
  		close();
  	}
  
  	public void loadFiles() throws IOException {
  		RawData.createSearchKeys(new Path(options.getWorkPath(), searchkeyf));
  		RawData.createUserAgents(new Path(options.getWorkPath(), uagentf));
  		RawData.createCCodes(new Path(options.getWorkPath(), countryf));
  	}
  
  	private void init() throws IOException {
  
  		log.info("Initializing hive date generator...");
  
  		Utils.checkHdfsPath(options.getResultPath(), true);
  		Utils.checkHdfsPath(options.getWorkPath(), true);
  
  		loadFiles();
  		
  		Utils.serialLinkZipf(options);
  		
  		dummy = new Dummy(options.getWorkPath(), options.getNumMaps());
  	}
  
  	public void close() throws IOException {
  
  		log.info("Closing hive data generator...");
  		Utils.checkHdfsPath(options.getWorkPath());
  	}
  }
  