code: package HiBench;
  
  import java.io.BufferedReader;
  import java.io.FileInputStream;
  import java.io.FileReader;
  import java.io.IOException;
  import java.io.ObjectInputStream;
  import java.io.ObjectOutputStream;
  import java.net.URI;
  import java.net.URISyntaxException;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.filecache.DistributedCache;
  import org.apache.hadoop.fs.FSDataOutputStream;
  import org.apache.hadoop.fs.FileStatus;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.fs.PathFilter;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.MapFile;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapred.ClusterStatus;
  import org.apache.hadoop.mapred.JobClient;
  import org.apache.hadoop.mapred.JobConf;
  
  public class Utils {
  
  	private static final Log log = LogFactory.getLog(Utils.class.getName());
  	
  	public static final void checkHdfsPath(Path path) throws IOException {
  		
  		FileSystem fs = path.getFileSystem(new Configuration());
  		if (fs.exists(path)) {
  			fs.delete(path, true);
  		}
  		fs.close();
  	}
  	
  	public static final void createHdfsDir(Path path) throws IOException {
  
  		FileSystem fs = path.getFileSystem(new Configuration());
  		if (!fs.exists(path)) {
  			fs.mkdirs(path);
  		}
  		fs.close();
  	}
  	
  	public static final void checkHdfsPath(Path path, boolean mkdir)
  			throws IOException {
  
  		FileSystem fs = path.getFileSystem(new Configuration());
  		if (fs.exists(path)) {
  			fs.delete(path, true);
  		}
  		
  		if (mkdir) {
  			fs.mkdirs(path);
  		}
  		fs.close();
  	}
  
  	public static final void cleanTempFiles (Path file) throws IOException {
  		Path ftemp = new Path(file, "_logs");
  		FileSystem fs = ftemp.getFileSystem(new Configuration());
  		fs.delete(ftemp, true);
  		fs.close();
  	}
  
  	public static final int getMaxNumReds () throws IOException {
  		JobConf job = new JobConf(Utils.class);
  		JobClient client = new JobClient(job);
  		ClusterStatus cluster = client.getClusterStatus();
  		return cluster.getMaxReduceTasks();
  	}
  	
  	public static final int getNumAvailableReds () throws IOException {
  		JobConf job = new JobConf(Utils.class);
  		JobClient client = new JobClient(job);
  		ClusterStatus cluster = client.getClusterStatus();
  		int maxReduces = cluster.getMaxReduceTasks();
  		int runnings = cluster.getReduceTasks();
  		return maxReduces - runnings;
  	}
  	
  	public static final int getMaxNumMaps () throws IOException {
  		JobConf job = new JobConf(Utils.class);
  		JobClient client = new JobClient(job);
  		ClusterStatus cluster = client.getClusterStatus();
  		return cluster.getMaxMapTasks();
  	}
  	
  	public static final int getNumAvailableMaps () throws IOException {
  		JobConf job = new JobConf(Utils.class);
  		JobClient client = new JobClient(job);
  		ClusterStatus cluster = client.getClusterStatus();
  		int maxMaps = cluster.getMaxMapTasks();
  		int runnings = cluster.getMapTasks();
  		return maxMaps - runnings;
  	}
  	
  	private static final void shareArray(String symbol, Path fdict, int words, JobConf job) throws URISyntaxException {
  		DistributedCache.createSymlink(job);
  		String uridict = fdict.toUri().toString() + "#" + symbol;
  		DistributedCache.addCacheFile(new URI(uridict), job);
  		job.setInt(symbol, words);
  	}
  	
  	private static final String[] getSharedArray(String symbol, JobConf job) throws IOException {
  		
  		String[] dict = null;
  		int words = job.getInt(symbol, 0);
  		if (words > 0) {
  			dict = new String[words];
  			
  			FileReader reader = new FileReader(symbol);
  			BufferedReader br = new BufferedReader(reader);
  			String line = null;
  			int len = 0;
  			while ((line = br.readLine()) != null) {
  				dict[len++] = line;
  			}
  			br.close();
  			reader.close();
  		}
  		return dict;
  	}
  	
  	public static final void shareDict(DataOptions options, JobConf job) throws URISyntaxException {
  		shareArray(HtmlCore.WORD_DICT_DIR_NAME, new Path(options.getWorkPath(), HtmlCore.WORD_DICT_DIR_NAME), options.getNumWords(), job);
  	}
  	
  	public static final String[] getDict(JobConf job) throws IOException {
  		return getSharedArray(HtmlCore.WORD_DICT_DIR_NAME, job);
  	}
  	
      private static PathFilter getPassDirectoriesFilter(final FileSystem fs) {
      	
          return new PathFilter() {
              public boolean accept(final Path path) {
                  try {
                      return fs.getFileStatus(path).isDir() &&
                              !path.getName().startsWith("_");
                      
                  } catch (IOException ioe) {
                      return false;
                  }
              }
          };
      }
      
  	private static final void shareMapFile(String symbol, int slots, Path mfile, JobConf job) throws IOException, URISyntaxException {
  		
  		FileSystem fs = FileSystem.get(mfile.toUri(), job);
  		if (fs.exists(mfile) && fs.getFileStatus(mfile).isDir()) {
  
  			DistributedCache.createSymlink(job);
  			
  			FileStatus[] fstats = fs.listStatus(mfile, getPassDirectoriesFilter(fs));
  			
  			LongWritable key = new LongWritable();
  			Text value = new Text();
  			for (int i=0; i<fstats.length; i++) {
  				Path curMap = fstats[i].getPath();
  				MapFile.Reader mreader = new MapFile.Reader(fs, curMap.toString(), job);
  				if (mreader.next(key, value)) {
  					int rid = (int) (key.get() % slots);
  					String uriWithLink =
  							curMap.toUri().toString() + "#" + symbol + "-" + Integer.toString(rid);
  					DistributedCache.addCacheFile(new URI(uriWithLink), job);
  				} else {
  					System.exit(-1);
  				}
  				mreader.close();
  			}
  		}
  		
  		job.setInt(symbol, slots);
  	}
  	
  	public static final void shareUrls (String symbol, DataOptions options, JobConf job) throws IOException, URISyntaxException {
  		shareMapFile(symbol, options.getNumMaps(), new Path(options.getWorkPath(), symbol), job);
  	}
  	
  	public static final IndexedMapFile getSharedMapFile(String symbol, JobConf job) throws IOException {
  		
  		int slots = job.getInt(symbol, 0);
  		
  		if (slots <=0) {
  			log.error("slots number should be no less than 1 !!!");
  			System.exit(-1);
  		}
  		
  		FileSystem fs = FileSystem.getLocal(job);
  		MapFile.Reader[] readers = new MapFile.Reader[slots];
  		for (int i=0; i<slots; i++) {
  			String symbfile = fs.getWorkingDirectory().toString() + "/" + symbol + "-" + Integer.toString(i);
  			readers[i] = new MapFile.Reader(fs, symbfile, job);
  		}
  		
  		return new IndexedMapFile(slots, readers);
  	}
  	
  	/***
  	 * Steps to make a ZipfCore available for each job
  	 * Client side
  	 * 		1. Zipfian creates its corresponding ZipfCore object
  	 * 		2. serialize the ZipfCore into a hdfs file
  	 * 		3. share the hdfs file by putting it into distributed cache file
  	 * Job side
  	 * 		1. read object from distributed cache file to re-create the ZipfCore
  	 * @throws IOException 
  	 */
  	private static final void serialZipfCore(Zipfian zipfian, Path hdfs_zipf) throws IOException {
  		
  		Utils.checkHdfsPath(hdfs_zipf);
  		
  		FileSystem fs = hdfs_zipf.getFileSystem(new Configuration());
  		FSDataOutputStream fout = fs.create(hdfs_zipf);
  
  		ObjectOutputStream so = new ObjectOutputStream(fout);
  		
  		ZipfCore core = zipfian.createZipfCore();
  		so.writeObject(core);
  		
  		so.close();
  		fout.close();
  		fs.close();
  	}
  	
  	private static final void shareZipfCore(String fname, DataOptions options, JobConf job) throws URISyntaxException {
  
  		DistributedCache.createSymlink(job);
  		
  		Path zipfPath = new Path(options.getWorkPath(), fname);
  		String uriWithLink = zipfPath.toString() + "#" + fname;
  		DistributedCache.addCacheFile(new URI(uriWithLink), job);
  	}
  	
  	private static final ZipfCore getSharedZipfCore(String fname, JobConf job) throws IOException, ClassNotFoundException {
  		
  		ZipfCore zipfcore = null;
  		
  		FileSystem fs = FileSystem.getLocal(job);
  		Path symbLink = new Path(fname);
  		if (fs.exists(symbLink)) {
  			FileInputStream fi = new FileInputStream(symbLink.toString());
  			ObjectInputStream si = new ObjectInputStream(fi);
  			
  			zipfcore = (ZipfCore) si.readObject();
  			si.close();
  		}
  		return zipfcore;
  	}
  	
  	public static final ZipfCore getSharedLinkZipfCore(JobConf job) throws ClassNotFoundException, IOException {
  		return getSharedZipfCore(HtmlCore.LINK_ZIPF_FILE_NAME, job);
  	}
  	
  	public static final ZipfCore getSharedWordZipfCore(JobConf job) throws ClassNotFoundException, IOException {
  		return getSharedZipfCore(HtmlCore.WORD_ZIPF_FILE_NAME, job);
  	}
  	
  	public static final void shareLinkZipfCore(DataOptions options, JobConf job) throws URISyntaxException {
  		shareZipfCore(HtmlCore.LINK_ZIPF_FILE_NAME, options, job);
  	}
  	
  	public static final void shareWordZipfCore(DataOptions options, JobConf job) throws URISyntaxException {
  		shareZipfCore(HtmlCore.WORD_ZIPF_FILE_NAME, options, job);
  	}
  	
  	public static final void serialLinkZipf(DataOptions options) throws IOException {
  		if (options.getNumPages() > 0) {
  			Zipfian lzipfian = new Zipfian(options.getNumPages(), DataOptions.LINK_ZIPF_EXPONENT);
  			lzipfian.setupZipf((long) (options.getNumPages() * HtmlCore.getMeanLinksPerPage()), DataOptions.LINK_SIMULATE_SPACE_RATIO);
  			Path lzipfCorePath = new Path(options.getWorkPath(), HtmlCore.LINK_ZIPF_FILE_NAME);
  			serialZipfCore(lzipfian, lzipfCorePath);
  		} else {
  			System.out.println("ERROR: number of pages should be greater than 0");
  			System.exit(-1);
  		}
  	}
  	
  	public static final void serialWordZipf(DataOptions options) throws IOException {
  		if (options.getNumWords() > 0) {
  			Zipfian wzipfian = new Zipfian(options.getNumWords(), DataOptions.WORD_ZIPF_EXPONENT);
  			wzipfian.setupZipf((long) (options.getNumPages() * HtmlCore.getMeanWordsPerPage()), DataOptions.WORD_SIMULATE_SPACE_RATIO);
  			Path wzipfCorePath = new Path(options.getWorkPath(), HtmlCore.WORD_ZIPF_FILE_NAME);
  			serialZipfCore(wzipfian, wzipfCorePath);
  		} else {
  			System.out.println("ERROR: number of words should be greater than 0");
  			System.exit(-1);
  		}
  	}
  }
  