MR总结(三)-MapReduce组件自定义

/**
 * Generate the list of files and make them into FileSplits.
 * @param job the job context
 * @throws IOException
 */
 public List getSplits(JobContext job) throws IOException {
 Stopwatch sw = new Stopwatch().start();
 long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
 long maxSize = getMaxSplitSize(job);

 // generate splits
 List splits = new ArrayList();
 List files = listStatus(job);
 for (FileStatus file: files) {
 Path path = file.getPath();
 long length = file.getLen();
 if (length != 0) {
 BlockLocation[] blkLocations;
 if (file instanceof LocatedFileStatus) {
 blkLocations = ((LocatedFileStatus) file).getBlockLocations();
 } else {
 FileSystem fs = path.getFileSystem(job.getConfiguration());
 blkLocations = fs.getFileBlockLocations(file, 0, length);
 }

 if (isSplitable(job, path)) {
 long blockSize = file.getBlockSize();
 long splitSize = computeSplitSize(blockSize, minSize, maxSize);

 long bytesRemaining = length;
 while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
 int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
 splits.add(makeSplit(path, length-bytesRemaining, splitSize,
 blkLocations[blkIndex].getHosts(),
 blkLocations[blkIndex].getCachedHosts()));
 bytesRemaining -= splitSize;
 }

 if (bytesRemaining != 0) {
 int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
 splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
 blkLocations[blkIndex].getHosts(),
 blkLocations[blkIndex].getCachedHosts()));
 }
 } else { // not splitable
 splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
 blkLocations[0].getCachedHosts()));
 }
 } else {
 //Create empty hosts array for zero length files
 splits.add(makeSplit(path, 0, length, new String[0]));
 }
 }
 // Save the number of input files for metrics/loadgen
 job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
 sw.stop();
 if (LOG.isDebugEnabled()) {
 LOG.debug("Total # of splits generated by getSplits: " + splits.size()
 + ", TimeTaken: " + sw.elapsedMillis());
 }
 return splits;
 }

//重写分片函数
 @Override
 public List getSplits(JobContext job) throws IOException {
 String[] servers = getActiveServersList(job);
 if (servers == null)
 return null;
 List splits = new ArrayList();
 List files = listStatus(job);
 int currentServer = 0;
 for (FileStatus file : files) {
 Path path = file.getPath();
 long length = file.getLen();
 if ((length != 0) && isSplitable(job, path)) {
 long blockSize = file.getBlockSize();
 long splitSize = computeSplitSize(blockSize, minSize, maxSize);
 long bytesRemaining = length;
 while (((double) bytesRemaining) / splitSize > SPLIT_SLOP) {
 splits.add(new FileSplit(path, length - bytesRemaining,
 splitSize, new String[] { servers[currentServer] }));
 currentServer = getNextServer(currentServer, servers.length);
 bytesRemaining -= splitSize;
 }
 } else if (length != 0) {
 splits.add(new FileSplit(path, 0, length,
 new String[] { servers[currentServer] }));
 currentServer = getNextServer(currentServer, servers.length);
 }
 }
 // Save the number of input files in the job-conf
 job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
 return splits;
 }
 //获取服务器列表
 private String[] getActiveServersList(JobContext context) {
 String[] servers = null;
 try {
 JobClient jc = new JobClient((JobConf) context.getConfiguration());
 ClusterStatus status = jc.getClusterStatus(true);
 Collection atc = status.getActiveTrackerNames();
 servers = new String[atc.size()];
 int s = 0;
 for (String serverInfo : atc) {
 StringTokenizer st = new StringTokenizer(serverInfo, ":");
 String trackerName = st.nextToken();
 StringTokenizer st1 = new StringTokenizer(trackerName, "_");
 st1.nextToken();
 servers[s++] = st1.nextToken();
 }
 } catch (IOException e) {
 e.printStackTrace();
 }
 return servers;
 }
 //选择一个服务器
 private static int getNextServer(int current, int max) {
 current++;
 if (current >= max)
 current = 0;
 return current;
 }

@Override
 public List getSplits(JobContext job) throws IOException {
 List originalSplits = super.getSplits(job);
 String[] servers = getActiveServersList(job);
 if (servers == null)
 return null;
 List splits = new ArrayList(
 originalSplits.size());
 int numSplits = originalSplits.size();
 int currentServer = 0;
 for (int i = 0; i < numSplits; i++, currentServer = getNextServer(
 currentServer, servers.length)) {
 String server = servers[currentServer]; // Current server
 boolean replaced = false;
 for (InputSplit split : originalSplits) {
 FileSplit fs = (FileSplit) split;
 for (String l : fs.getLocations()) {
 if (l.equals(server)) {
 splits.add(new FileSplit(fs.getPath(), fs.getStart(),
 fs.getLength(), new String[] { server }));
 originalSplits.remove(split);
 replaced = true;
 break;
 }
 }
 if (replaced)
 break;
 }
 if (!replaced) {
 FileSplit fs = (FileSplit) splits.get(0);
 splits.add(new FileSplit(fs.getPath(), fs.getStart(), fs
 .getLength(), new String[] { server }));
 originalSplits.remove(0);
 }
 }
 return splits;
 }