src/contrib/gridmix/src/test/org/apache/hadoop/mapred/gridmix/DebugJobFactory.java - hadoop-mapreduce - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.hadoop.mapred.gridmix;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Random;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.JobID;
 import org.apache.hadoop.mapreduce.TaskType;
 import org.apache.hadoop.tools.rumen.JobStory;
 import org.apache.hadoop.tools.rumen.JobStoryProducer;
 import org.apache.hadoop.tools.rumen.TaskAttemptInfo;
 import org.apache.hadoop.tools.rumen.TaskInfo;
 import org.apache.hadoop.tools.rumen.Pre21JobHistoryConstants.Values;;


 /**
  * Component generating random job traces for testing on a single node.
  */
 class DebugJobFactory extends JobFactory {

   public DebugJobFactory(JobSubmitter submitter, Path scratch, int numJobs,
       Configuration conf, CountDownLatch startFlag) throws IOException {
     super(submitter, new DebugJobProducer(numJobs, conf), scratch, conf,
         startFlag);
   }

   ArrayList<JobStory> getSubmitted() {
     return ((DebugJobProducer)jobProducer).submitted;
   }

   private static class DebugJobProducer implements JobStoryProducer {
     final ArrayList<JobStory> submitted;
     private final Configuration conf;
     private final AtomicInteger numJobs;

     public DebugJobProducer(int numJobs, Configuration conf) {
       super();
       this.conf = conf;
       this.numJobs = new AtomicInteger(numJobs);
       this.submitted = new ArrayList<JobStory>();
     }

     @Override
     public JobStory getNextJob() throws IOException {
       if (numJobs.getAndDecrement() > 0) {
         final MockJob ret = new MockJob(conf);
         submitted.add(ret);
         return ret;
       }
       return null;
     }

     @Override
     public void close() { }
   }

   static double[] getDistr(Random r, double mindist, int size) {
     assert 0.0 <= mindist && mindist <= 1.0;
     final double min = mindist / size;
     final double rem = 1.0 - min * size;
     final double[] tmp = new double[size];
     for (int i = 0; i < tmp.length - 1; ++i) {
       tmp[i] = r.nextDouble() * rem;
     }
     tmp[tmp.length - 1] = rem;
     Arrays.sort(tmp);

     final double[] ret = new double[size];
     ret[0] = tmp[0] + min;
     for (int i = 1; i < size; ++i) {
       ret[i] = tmp[i] - tmp[i-1] + min;
     }
     return ret;
   }

   /**
    * Generate random task data for a synthetic job.
    */
   static class MockJob implements JobStory {

     static final int MIN_REC = 1 << 14;
     static final int MIN_BYTES = 1 << 20;
     static final int VAR_REC = 1 << 14;
     static final int VAR_BYTES = 4 << 20;
     static final int MAX_MAP = 5;
     static final int MAX_RED = 3;

     static void initDist(Random r, double min, int[] recs, long[] bytes,
         long tot_recs, long tot_bytes) {
       final double[] recs_dist = getDistr(r, min, recs.length);
       final double[] bytes_dist = getDistr(r, min, recs.length);
       long totalbytes = 0L;
       int totalrecs = 0;
       for (int i = 0; i < recs.length; ++i) {
         recs[i] = (int) Math.round(tot_recs * recs_dist[i]);
         bytes[i] = Math.round(tot_bytes * bytes_dist[i]);
         totalrecs += recs[i];
         totalbytes += bytes[i];
       }
       // Add/remove excess
       recs[0] += totalrecs - tot_recs;
       bytes[0] += totalbytes - tot_bytes;
       if (LOG.isInfoEnabled()) {
         LOG.info("DIST: " + Arrays.toString(recs) + " " +
             tot_recs + "/" + totalrecs + " " +
             Arrays.toString(bytes) + " " + tot_bytes + "/" + totalbytes);
       }
     }

     private static final AtomicInteger seq = new AtomicInteger(0);
     // set timestamps in the past
     private static final AtomicLong timestamp =
       new AtomicLong(System.currentTimeMillis() -
         TimeUnit.MILLISECONDS.convert(60, TimeUnit.DAYS));

     private final int id;
     private final String name;
     private final int[] m_recsIn, m_recsOut, r_recsIn, r_recsOut;
     private final long[] m_bytesIn, m_bytesOut, r_bytesIn, r_bytesOut;
     private final long submitTime;

     public MockJob(Configuration conf) {
       final Random r = new Random();
       final long seed = r.nextLong();
       r.setSeed(seed);
       id = seq.getAndIncrement();
       name = String.format("MOCKJOB%05d", id);
       LOG.info(name + " (" + seed + ")");
       submitTime = timestamp.addAndGet(TimeUnit.MILLISECONDS.convert(
             r.nextInt(10), TimeUnit.SECONDS));

       m_recsIn = new int[r.nextInt(MAX_MAP) + 1];
       m_bytesIn = new long[m_recsIn.length];
       m_recsOut = new int[m_recsIn.length];
       m_bytesOut = new long[m_recsIn.length];

       r_recsIn = new int[r.nextInt(MAX_RED) + 1];
       r_bytesIn = new long[r_recsIn.length];
       r_recsOut = new int[r_recsIn.length];
       r_bytesOut = new long[r_recsIn.length];

       // map input
       final long map_recs = r.nextInt(VAR_REC) + MIN_REC;
       final long map_bytes = r.nextInt(VAR_BYTES) + MIN_BYTES;
       initDist(r, 0.5, m_recsIn, m_bytesIn, map_recs, map_bytes);

       // shuffle
       final long shuffle_recs = r.nextInt(VAR_REC) + MIN_REC;
       final long shuffle_bytes = r.nextInt(VAR_BYTES) + MIN_BYTES;
       initDist(r, 0.4, m_recsOut, m_bytesOut, shuffle_recs, shuffle_bytes);
       initDist(r, 0.8, r_recsIn, r_bytesIn, shuffle_recs, shuffle_bytes);

       // reduce output
       final long red_recs = r.nextInt(VAR_REC) + MIN_REC;
       final long red_bytes = r.nextInt(VAR_BYTES) + MIN_BYTES;
       initDist(r, 0.4, r_recsOut, r_bytesOut, red_recs, red_bytes);

       if (LOG.isDebugEnabled()) {
         int iMapBTotal = 0, oMapBTotal = 0, iRedBTotal = 0, oRedBTotal = 0;
         int iMapRTotal = 0, oMapRTotal = 0, iRedRTotal = 0, oRedRTotal = 0;
         for (int i = 0; i < m_recsIn.length; ++i) {
           iMapRTotal += m_recsIn[i];
           iMapBTotal += m_bytesIn[i];
           oMapRTotal += m_recsOut[i];
           oMapBTotal += m_bytesOut[i];
         }
         for (int i = 0; i < r_recsIn.length; ++i) {
           iRedRTotal += r_recsIn[i];
           iRedBTotal += r_bytesIn[i];
           oRedRTotal += r_recsOut[i];
           oRedBTotal += r_bytesOut[i];
         }
         LOG.debug(String.format("%s: M (%03d) %6d/%10d -> %6d/%10d" +
                                    " R (%03d) %6d/%10d -> %6d/%10d @%d", name,
             m_bytesIn.length, iMapRTotal, iMapBTotal, oMapRTotal, oMapBTotal,
             r_bytesIn.length, iRedRTotal, iRedBTotal, oRedRTotal, oRedBTotal,
             submitTime));
       }
     }

     @Override
     public String getName() {
       return name;
     }

     @Override
     public String getUser() {
       return "FOOBAR";
     }

     @Override
     public JobID getJobID() {
       return new JobID("job_mock_" + name, id);
     }

     @Override
     public Values getOutcome() {
       return Values.SUCCESS;
     }

     @Override
     public long getSubmissionTime() {
       return submitTime;
     }

     @Override
     public int getNumberMaps() {
       return m_bytesIn.length;
     }

     @Override
     public int getNumberReduces() {
       return r_bytesIn.length;
     }

     @Override
     public TaskInfo getTaskInfo(TaskType taskType, int taskNumber) {
       switch (taskType) {
         case MAP:
           return new TaskInfo(m_bytesIn[taskNumber], m_recsIn[taskNumber],
               m_bytesOut[taskNumber], m_recsOut[taskNumber], -1);
         case REDUCE:
           return new TaskInfo(r_bytesIn[taskNumber], r_recsIn[taskNumber],
               r_bytesOut[taskNumber], r_recsOut[taskNumber], -1);
         default:
           throw new IllegalArgumentException("Not interested");
       }
     }

     @Override
     public InputSplit[] getInputSplits() {
       throw new UnsupportedOperationException();
     }

     @Override
     public TaskAttemptInfo getTaskAttemptInfo(TaskType taskType,
         int taskNumber, int taskAttemptNumber) {
       throw new UnsupportedOperationException();
     }

     @Override
     public TaskAttemptInfo getMapTaskAttemptInfoAdjusted(int taskNumber,
         int taskAttemptNumber, int locality) {
       throw new UnsupportedOperationException();
     }

     @Override
     public org.apache.hadoop.mapred.JobConf getJobConf() {
       throw new UnsupportedOperationException();
     }
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.hadoop.mapred.gridmix;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Random;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicLong;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.mapreduce.InputSplit;
	import org.apache.hadoop.mapreduce.JobID;
	import org.apache.hadoop.mapreduce.TaskType;
	import org.apache.hadoop.tools.rumen.JobStory;
	import org.apache.hadoop.tools.rumen.JobStoryProducer;
	import org.apache.hadoop.tools.rumen.TaskAttemptInfo;
	import org.apache.hadoop.tools.rumen.TaskInfo;
	import org.apache.hadoop.tools.rumen.Pre21JobHistoryConstants.Values;;


	/**
	* Component generating random job traces for testing on a single node.
	*/
	class DebugJobFactory extends JobFactory {

	public DebugJobFactory(JobSubmitter submitter, Path scratch, int numJobs,
	Configuration conf, CountDownLatch startFlag) throws IOException {
	super(submitter, new DebugJobProducer(numJobs, conf), scratch, conf,
	startFlag);
	}

	ArrayList<JobStory> getSubmitted() {
	return ((DebugJobProducer)jobProducer).submitted;
	}

	private static class DebugJobProducer implements JobStoryProducer {
	final ArrayList<JobStory> submitted;
	private final Configuration conf;
	private final AtomicInteger numJobs;

	public DebugJobProducer(int numJobs, Configuration conf) {
	super();
	this.conf = conf;
	this.numJobs = new AtomicInteger(numJobs);
	this.submitted = new ArrayList<JobStory>();
	}

	@Override
	public JobStory getNextJob() throws IOException {
	if (numJobs.getAndDecrement() > 0) {
	final MockJob ret = new MockJob(conf);
	submitted.add(ret);
	return ret;
	}
	return null;
	}

	@Override
	public void close() { }
	}

	static double[] getDistr(Random r, double mindist, int size) {
	assert 0.0 <= mindist && mindist <= 1.0;
	final double min = mindist / size;
	final double rem = 1.0 - min * size;
	final double[] tmp = new double[size];
	for (int i = 0; i < tmp.length - 1; ++i) {
	tmp[i] = r.nextDouble() * rem;
	}
	tmp[tmp.length - 1] = rem;
	Arrays.sort(tmp);

	final double[] ret = new double[size];
	ret[0] = tmp[0] + min;
	for (int i = 1; i < size; ++i) {
	ret[i] = tmp[i] - tmp[i-1] + min;
	}
	return ret;
	}

	/**
	* Generate random task data for a synthetic job.
	*/
	static class MockJob implements JobStory {

	static final int MIN_REC = 1 << 14;
	static final int MIN_BYTES = 1 << 20;
	static final int VAR_REC = 1 << 14;
	static final int VAR_BYTES = 4 << 20;
	static final int MAX_MAP = 5;
	static final int MAX_RED = 3;

	static void initDist(Random r, double min, int[] recs, long[] bytes,
	long tot_recs, long tot_bytes) {
	final double[] recs_dist = getDistr(r, min, recs.length);
	final double[] bytes_dist = getDistr(r, min, recs.length);
	long totalbytes = 0L;
	int totalrecs = 0;
	for (int i = 0; i < recs.length; ++i) {
	recs[i] = (int) Math.round(tot_recs * recs_dist[i]);
	bytes[i] = Math.round(tot_bytes * bytes_dist[i]);
	totalrecs += recs[i];
	totalbytes += bytes[i];
	}
	// Add/remove excess
	recs[0] += totalrecs - tot_recs;
	bytes[0] += totalbytes - tot_bytes;
	if (LOG.isInfoEnabled()) {
	LOG.info("DIST: " + Arrays.toString(recs) + " " +
	tot_recs + "/" + totalrecs + " " +
	Arrays.toString(bytes) + " " + tot_bytes + "/" + totalbytes);
	}
	}

	private static final AtomicInteger seq = new AtomicInteger(0);
	// set timestamps in the past
	private static final AtomicLong timestamp =
	new AtomicLong(System.currentTimeMillis() -
	TimeUnit.MILLISECONDS.convert(60, TimeUnit.DAYS));

	private final int id;
	private final String name;
	private final int[] m_recsIn, m_recsOut, r_recsIn, r_recsOut;
	private final long[] m_bytesIn, m_bytesOut, r_bytesIn, r_bytesOut;
	private final long submitTime;

	public MockJob(Configuration conf) {
	final Random r = new Random();
	final long seed = r.nextLong();
	r.setSeed(seed);
	id = seq.getAndIncrement();
	name = String.format("MOCKJOB%05d", id);
	LOG.info(name + " (" + seed + ")");
	submitTime = timestamp.addAndGet(TimeUnit.MILLISECONDS.convert(
	r.nextInt(10), TimeUnit.SECONDS));

	m_recsIn = new int[r.nextInt(MAX_MAP) + 1];
	m_bytesIn = new long[m_recsIn.length];
	m_recsOut = new int[m_recsIn.length];
	m_bytesOut = new long[m_recsIn.length];

	r_recsIn = new int[r.nextInt(MAX_RED) + 1];
	r_bytesIn = new long[r_recsIn.length];
	r_recsOut = new int[r_recsIn.length];
	r_bytesOut = new long[r_recsIn.length];

	// map input
	final long map_recs = r.nextInt(VAR_REC) + MIN_REC;
	final long map_bytes = r.nextInt(VAR_BYTES) + MIN_BYTES;
	initDist(r, 0.5, m_recsIn, m_bytesIn, map_recs, map_bytes);

	// shuffle
	final long shuffle_recs = r.nextInt(VAR_REC) + MIN_REC;
	final long shuffle_bytes = r.nextInt(VAR_BYTES) + MIN_BYTES;
	initDist(r, 0.4, m_recsOut, m_bytesOut, shuffle_recs, shuffle_bytes);
	initDist(r, 0.8, r_recsIn, r_bytesIn, shuffle_recs, shuffle_bytes);

	// reduce output
	final long red_recs = r.nextInt(VAR_REC) + MIN_REC;
	final long red_bytes = r.nextInt(VAR_BYTES) + MIN_BYTES;
	initDist(r, 0.4, r_recsOut, r_bytesOut, red_recs, red_bytes);

	if (LOG.isDebugEnabled()) {
	int iMapBTotal = 0, oMapBTotal = 0, iRedBTotal = 0, oRedBTotal = 0;
	int iMapRTotal = 0, oMapRTotal = 0, iRedRTotal = 0, oRedRTotal = 0;
	for (int i = 0; i < m_recsIn.length; ++i) {
	iMapRTotal += m_recsIn[i];
	iMapBTotal += m_bytesIn[i];
	oMapRTotal += m_recsOut[i];
	oMapBTotal += m_bytesOut[i];
	}
	for (int i = 0; i < r_recsIn.length; ++i) {
	iRedRTotal += r_recsIn[i];
	iRedBTotal += r_bytesIn[i];
	oRedRTotal += r_recsOut[i];
	oRedBTotal += r_bytesOut[i];
	}
	LOG.debug(String.format("%s: M (%03d) %6d/%10d -> %6d/%10d" +
	" R (%03d) %6d/%10d -> %6d/%10d @%d", name,
	m_bytesIn.length, iMapRTotal, iMapBTotal, oMapRTotal, oMapBTotal,
	r_bytesIn.length, iRedRTotal, iRedBTotal, oRedRTotal, oRedBTotal,
	submitTime));
	}
	}

	@Override
	public String getName() {
	return name;
	}

	@Override
	public String getUser() {
	return "FOOBAR";
	}

	@Override
	public JobID getJobID() {
	return new JobID("job_mock_" + name, id);
	}

	@Override
	public Values getOutcome() {
	return Values.SUCCESS;
	}

	@Override
	public long getSubmissionTime() {
	return submitTime;
	}

	@Override
	public int getNumberMaps() {
	return m_bytesIn.length;
	}

	@Override
	public int getNumberReduces() {
	return r_bytesIn.length;
	}

	@Override
	public TaskInfo getTaskInfo(TaskType taskType, int taskNumber) {
	switch (taskType) {
	case MAP:
	return new TaskInfo(m_bytesIn[taskNumber], m_recsIn[taskNumber],
	m_bytesOut[taskNumber], m_recsOut[taskNumber], -1);
	case REDUCE:
	return new TaskInfo(r_bytesIn[taskNumber], r_recsIn[taskNumber],
	r_bytesOut[taskNumber], r_recsOut[taskNumber], -1);
	default:
	throw new IllegalArgumentException("Not interested");
	}
	}

	@Override
	public InputSplit[] getInputSplits() {
	throw new UnsupportedOperationException();
	}

	@Override
	public TaskAttemptInfo getTaskAttemptInfo(TaskType taskType,
	int taskNumber, int taskAttemptNumber) {
	throw new UnsupportedOperationException();
	}

	@Override
	public TaskAttemptInfo getMapTaskAttemptInfoAdjusted(int taskNumber,
	int taskAttemptNumber, int locality) {
	throw new UnsupportedOperationException();
	}

	@Override
	public org.apache.hadoop.mapred.JobConf getJobConf() {
	throw new UnsupportedOperationException();
	}
	}
	}