code: /**
   * 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.fs.dfsioe;
  
  import java.io.IOException;
  import java.util.StringTokenizer;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.fs.dfsioe.TestDFSIOEnh.LinearInterpolator;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  
  public class Analyzer {
  	
  	public static class _Mapper extends Mapper<Object, Text, Text, Text> {
  		private Text t = new Text();
  		
  		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
  			StringTokenizer tokens = new StringTokenizer(value.toString(), " \t\n\r\f%");
  			String attr = tokens.nextToken(); 
  			if (attr.endsWith(":tput_samples")){
  	            String[] tags=attr.split(":");
  	            String[] samples = tokens.nextToken().split(";");
  				for(int j=0; !samples[j].startsWith("EoR"); j++){
  					t.set(samples[j]);
  	                context.write(new Text(tags[1]), t);
  	            }
  	        }
  		}
  	}
  	
  	public static class _Reducer extends Reducer<Text, Text, Text, Text> {
  		private Text result = new Text();
  		private int plotInterval;
  		private long sampleUnit;
  		private long execTime;
  		private long fileSize;
  		private long tStart;
  		private int maxslot;
  				
  		public void setup(Context context) throws IOException, InterruptedException {
  			this.tStart = context.getConfiguration().getLong("ana_tStart", 1347347421736L);
  			this.plotInterval = context.getConfiguration().getInt("ana_plotInterval", 1000);
  			this.sampleUnit = context.getConfiguration().getLong("ana_sampleUnit", 1024*1024);
  			this.execTime = context.getConfiguration().getLong("ana_execTime", 767829);
  			this.fileSize = context.getConfiguration().getLong("ana_fileSize", 500*1024*1024);
  			this.maxslot = (int)(execTime/plotInterval)+1;
  		}
  		
  		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
  	        LinearInterpolator processor = new LinearInterpolator();
  			for (Text val : values) {
  				String[] ss = val.toString().split(":");
  				long timeStamp = Long.parseLong(ss[0]);
                  long bytes = Long.parseLong(ss[1]);    
                  double timePassed = (timeStamp-tStart)/(double)plotInterval;
                  processor.add((double)timePassed,(double)bytes);
  			}
  			processor.add(0, 0);
  			processor.add(maxslot+0.1, fileSize);
  			double[] resultValue = new double [maxslot+1];
  			double[]  bytesChanged = new double[maxslot+1];
  			for(int i = 0; i<=maxslot; i++){
  	            resultValue[i]=processor.get(i);
  		    }
  			for (int i = 0; i<=maxslot-1; i++) {
                  bytesChanged[i] = resultValue[i+1]-resultValue[i];
  			}
  			bytesChanged[maxslot] = 0;
  			for (int ri = 0; ri<=maxslot; ri++) {
  				result.set(ri+","+resultValue[ri]/(double)sampleUnit+","+bytesChanged[ri]/(double)sampleUnit);
                  context.write(key, result);
  			}
  		}
  	}
  	
  	public static void main(String[] args) throws Exception{
  	    Configuration conf = new Configuration();
  	    FileSystem fs = FileSystem.get(conf);
  	    Path outdir = new Path("/result.txt");
  	    fs.delete(outdir, true);
  	    Job job = new Job(conf, "Result Analyzer");
  	    job.setJarByClass(Analyzer.class);
  	    job.setMapperClass(_Mapper.class);
  	    job.setReducerClass(_Reducer.class);
  	    job.setOutputKeyClass(Text.class);
  	    job.setOutputValueClass(Text.class);
  	    FileInputFormat.addInputPath(job, new Path(DfsioeConfig.getInstance().getReadDir(conf), "part-00000"));
  	    FileOutputFormat.setOutputPath(job, outdir);
  	    System.exit(job.waitForCompletion(true) ? 0 : 1);
  	}
  }