import org.apache.hadoop.vaidya.statistics.job.MapTaskStatistics; //导入依赖的package包/类
@Override
public double evaluate(JobStatistics job) {

  /*
   * Set the this._job
   */
  this._job = job;
    
  /*
   * Read the Normalization Factor
   */
  double normF = getInputElementDoubleValue("NormalizationFactor", 3.0);
  
  /*
   * Get the sorted map task list by number MapTaskKeys.OUTPUT_BYTES
   */
  List<MapTaskStatistics> smTaskList = job.getMapTaskList(MapTaskKeys.FILE_BYTES_WRITTEN, KeyDataType.LONG);
  int size = smTaskList.size();
  long numLocalBytesWrittenByMaps = 0;
  for (int i=0; i<size; i++) {
    numLocalBytesWrittenByMaps += smTaskList.get(i).getLongValue(MapTaskKeys.FILE_BYTES_WRITTEN);
  }
  this._numLocalBytesWrittenByMaps = numLocalBytesWrittenByMaps;
  
  /*
   * Map only job vs. map reduce job
   * For MapReduce job MAP_OUTPUT_BYTES are normally written by maps on local disk, so they are subtracted
   * from the localBytesWrittenByMaps.
   */
  if (job.getLongValue(JobKeys.TOTAL_REDUCES) > 0) {
    this._impact = (this._numLocalBytesWrittenByMaps - job.getLongValue(JobKeys.MAP_OUTPUT_BYTES))/job.getLongValue(JobKeys.MAP_OUTPUT_BYTES);
  } else {
    this._impact = this._numLocalBytesWrittenByMaps/job.getLongValue(JobKeys.MAP_OUTPUT_BYTES);
  }
  
  if (this._impact > normF) {
    this._impact = 1.0;
  } else {
    this._impact = this._impact/normF;
  }
  
  return this._impact;
  
}
 

import org.apache.hadoop.vaidya.statistics.job.MapTaskStatistics; //导入依赖的package包/类
@Override
public double evaluate(JobStatistics job) {

  /*
   * Set the this._job
   */
  this._job = job;
    
  /*
   * Read the Normalization Factor
   */
  double normF = getInputElementDoubleValue("NormalizationFactor", 3.0);
  
  /*
   * Get the sorted reduce task list by number MapTaskKeys.OUTPUT_BYTES
   */
  List<MapTaskStatistics> srTaskList = job.getMapTaskList(MapTaskKeys.LOCAL_BYTES_WRITTEN, KeyDataType.LONG);
  int size = srTaskList.size();
  long numLocalBytesWrittenByMaps = 0;
  for (int i=0; i<size; i++) {
    numLocalBytesWrittenByMaps += srTaskList.get(i).getLongValue(MapTaskKeys.LOCAL_BYTES_WRITTEN);
  }
  this._numLocalBytesWrittenByMaps = numLocalBytesWrittenByMaps;
  
  /*
   * Map only job vs. map reduce job
   */
  if (job.getLongValue(JobKeys.TOTAL_REDUCES) > 0) {
    this._impact = (this._numLocalBytesWrittenByMaps - job.getLongValue(JobKeys.MAP_OUTPUT_BYTES))/job.getLongValue(JobKeys.MAP_OUTPUT_BYTES);
  } else {
    this._impact = this._numLocalBytesWrittenByMaps/job.getLongValue(JobKeys.MAP_OUTPUT_BYTES);
  }
  
  if (this._impact > normF) {
    this._impact = 1.0;
  } else {
    this._impact = this._impact/normF;
  }
  
  return this._impact;
  
}