Get Started Analyzing Data with Hadoop

MapReduce works by breaking the processing into two phases: the map phase and the reduce phase. Each phase has key-value pairs as input and output, the types of which may be chosen by the programmer. The programmer also specifies two functions: the map function and the reduce function.

The input to our map phase is the raw NCDC data. We choose a text input format that gives us each line in the dataset as a text value. The key is the offset of the beginning of the line from the beginning of the file, but as we have no need for this, we ignore it.

Our map function is simple. We pull out the year and the air temperature, since these are the only fields we are interested in. In this case, the map function is just a data preparation phase, setting up the data in such a way that the reducer function can do its work on it: finding the maximum temperature for each year. The map function is also a good place to drop bad records: here we filter out temperatures that are missing, suspect, or erroneous.

To visualize the way the map works, consider the following sample lines of input data (some unused columns have been dropped to fit the page, indicated by ellipses):

0067011990999991950051507004...9999999N9+00001+99999999999...

0043011990999991950051512004...9999999N9+00221+99999999999...

0043011990999991950051518004...9999999N9-00111+99999999999...

0043012650999991949032412004...0500001N9+01111+99999999999...

0043012650999991949032418004...0500001N9+00781+99999999999...

These lines are presented to the map function as the key-value pairs:

(0, 0067011990999991950051507004...9999999N9+00001+99999999999...)

(106, 0043011990999991950051512004...9999999N9+00221+99999999999...)

(212, 0043011990999991950051518004...9999999N9-00111+99999999999...)

(318, 0043012650999991949032412004...0500001N9+01111+99999999999...)

(424, 0043012650999991949032418004...0500001N9+00781+99999999999...)

The keys are the line offsets within the file, which we ignore in our map function. The map function merely extracts the year and the air temperature (indicated in bold text), and emits them as its output. (The temperature values have been interpreted as integers.)

(1950, 0)

(1950, 22)

(1950, −11)

(1949, 111)

(1949, 78)

The output from the map function is processed by the MapReduce framework before being sent to the reduce function. This processing sorts and groups the key-value pairs by key. So, continuing the example, our reduce function sees the following input:

(1949, [111, 78])

(1950, [0, 22, −11])

Each year appears with a list of all its air temperature readings. All the reduce function has to do now is iterate through the list and pick up the maximum reading:

(1949, 111)

(1950, 22)

This is the final output: the maximum global temperature recorded in each year.

The whole data flow is illustrated in Figure 2.1 At the bottom of the diagram is a Unix pipeline, which mimics the whole MapReduce flow, and which we will see again later in the chapter when we look at Hadoop Streaming.

Figure 2.1. MapReduce logical data flow

Having run through how the MapReduce program works, the next step is to express it in code. We need three things: a map function, a reduce function, and some code to run the job. The map function is represented by an implementation of the Mapper interface, which declares a map() method. Example 2.3 shows the implementation of our map function.

import java.io.IOException;



import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.MapReduceBase;

import org.apache.hadoop.mapred.Mapper;

import org.apache.hadoop.mapred.OutputCollector;

import org.apache.hadoop.mapred.Reporter;



public class MaxTemperatureMapper extends MapReduceBase

  implements Mapper {



  private static final int MISSING = 9999;

  

  public void map(LongWritable key, Text value,

      OutputCollector output, Reporter reporter)

      throws IOException {

    

    String line = value.toString();

    String year = line.substring(15, 19);

    int airTemperature;

    if (line.charAt(87) == '+') { // parseInt doesn't like leading plus signs

      airTemperature = Integer.parseInt(line.substring(88, 92));

    } else {

      airTemperature = Integer.parseInt(line.substring(87, 92));

    }

    String quality = line.substring(92, 93);

    if (airTemperature != MISSING && quality.matches("[01459]")) {

      output.collect(new Text(year), new IntWritable(airTemperature));

    }

  }

}

The Mapper interface is a generic type, with four formal type parameters that specify the input key, input value, output key, and output value types of the map function. For the present example, the input key is a long integer offset, the input value is a line of text, the output key is a year, and the output value is an air temperature (an integer). Rather than use built-in Java types, Hadoop provides its own set of basic types that are optimized for network serialization. These are found in the org.apache.hadoop.io package. Here we use LongWritable, which corresponds to a Java Long, Text (like Java String), and IntWritable (like Java Integer).

The map() method is passed a key and a value. We convert the Text value containing the line of input into a Java String, then use its substring() method to extract the columns we are interested in.

The map() method also provides an instance of OutputCollector to write the output to. In this case, we write the year as a Text object (since we are just using it as a key), and the temperature wrapped in an IntWritable. We write an output record only if the temperature is present and the quality code indicates the temperature reading is OK.

The reduce function is similarly defined using a Reducer, as illustrated in Example 2.4.

import java.io.IOException;

import java.util.Iterator;



import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.MapReduceBase;

import org.apache.hadoop.mapred.OutputCollector;

import org.apache.hadoop.mapred.Reducer;

import org.apache.hadoop.mapred.Reporter;



public class MaxTemperatureReducer extends MapReduceBase

  implements Reducer {



  public void reduce(Text key, Iterator values,

      OutputCollector output, Reporter reporter)

      throws IOException {

    

    int maxValue = Integer.MIN_VALUE;

    while (values.hasNext()) {

      maxValue = Math.max(maxValue, values.next().get());

    }

    output.collect(key, new IntWritable(maxValue));

  }

}

Again, four formal type parameters are used to specify the input and output types, this time for the reduce function. The input types of the reduce function must match the output type of the map function: Text and IntWritable. And in this case, the output types of the reduce function are Text and IntWritable, for a year and its maximum temperature, which we find by iterating through the temperatures and comparing each with a record of the highest found so far.

The third piece of code runs the MapReduce job (see Example 2.5).

import java.io.IOException;



import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.FileInputFormat;

import org.apache.hadoop.mapred.FileOutputFormat;

import org.apache.hadoop.mapred.JobClient;

import org.apache.hadoop.mapred.JobConf;



public class MaxTemperature {



  public static void main(String[] args) throws IOException {

    if (args.length != 2) {

      System.err.println("Usage: MaxTemperature  
");

      System.exit(-1);

    }

    

    JobConf conf = new JobConf(MaxTemperature.class);

    conf.setJobName("Max temperature");



    FileInputFormat.addInputPath(conf, new Path(args[0]));

    FileOutputFormat.setOutputPath(conf, new Path(args[1]));

    

    conf.setMapperClass(MaxTemperatureMapper.class);

    conf.setReducerClass(MaxTemperatureReducer.class);



    conf.setOutputKeyClass(Text.class);

    conf.setOutputValueClass(IntWritable.class);



    JobClient.runJob(conf);

  }

}

A JobConf object forms the specification of the job. It gives you control over how the job is run. When we run this job on a Hadoop cluster, we will package the code into a JAR file (which Hadoop will distribute round the cluster). Rather than explicitly specify the name of the JAR file, we can pass a class in the JobConf constructor, which Hadoop will use to locate the relevant JAR file by looking for the JAR file containing this class.

Having constructed a JobConf object, we specify the input and output paths. An input path is specified by calling the static addInputPath() method on FileInputFormat, and it can be a single file, a directory (in which case, the input forms all the files in that directory), or a file pattern. As the name suggests, addInputPath() can be called more than once to use input from multiple paths.

The output path (of which there is only one) is specified by the static setOutputPath() method on FileOutputFormat. It specifies a directory where the output files from the reducer functions are written. The directory shouldn’t exist before running the job, as Hadoop will complain and not run the job. This precaution is to prevent data loss (it can be very annoying to accidentally overwrite the output of a long job with another).

Next, we specify the map and reduce types to use via the setMapperClass() and setReducerClass() methods.

The setOutputKeyClass() and setOutputValueClass() methods control the output types for the map and the reduce functions, which are often the same, as they are in our case. If they are different, then the map output types can be set using the methods setMapOutputKeyClass() and setMapOutputValueClass().

The input types are controlled via the input format, which we have not explicitly set since we are using the default TextInputFormat.

After setting the classes that define the map and reduce functions, we are ready to run the job. The static runJob() method on JobClient submits the job and waits for it to finish, writing information about its progress to the console.

% export HADOOP_CLASSPATH=build/classes

% hadoop MaxTemperature input/ncdc/sample.txt output

09/04/07 12:34:35 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName=Job

Tracker, sessionId=

09/04/07 12:34:35 WARN mapred.JobClient: Use GenericOptionsParser for parsing the 

arguments. Applications should implement Tool for the same.

09/04/07 12:34:35 WARN mapred.JobClient: No job jar file set.  User classes may not 

be found. See JobConf(Class) or JobConf#setJar(String).

09/04/07 12:34:35 INFO mapred.FileInputFormat: Total input paths to process : 1

09/04/07 12:34:35 INFO mapred.JobClient: Running job: job_local_0001

09/04/07 12:34:35 INFO mapred.FileInputFormat: Total input paths to process : 1

09/04/07 12:34:35 INFO mapred.MapTask: numReduceTasks: 1

09/04/07 12:34:35 INFO mapred.MapTask: io.sort.mb = 100

09/04/07 12:34:35 INFO mapred.MapTask: data buffer = 79691776/99614720

09/04/07 12:34:35 INFO mapred.MapTask: record buffer = 262144/327680

09/04/07 12:34:35 INFO mapred.MapTask: Starting flush of map output

09/04/07 12:34:36 INFO mapred.MapTask: Finished spill 0

09/04/07 12:34:36 INFO mapred.TaskRunner: Task:attempt_local_0001_m_000000_0 is 

done. And is in the process of commiting

09/04/07 12:34:36 INFO mapred.LocalJobRunner: file:/Users/tom/workspace/htdg/input/n

cdc/sample.txt:0+529

09/04/07 12:34:36 INFO mapred.TaskRunner: Task 'attempt_local_0001_m_000000_0' done.

09/04/07 12:34:36 INFO mapred.LocalJobRunner: 

09/04/07 12:34:36 INFO mapred.Merger: Merging 1 sorted segments

09/04/07 12:34:36 INFO mapred.Merger: Down to the last merge-pass, with 1 segments 

left of total size: 57 bytes

09/04/07 12:34:36 INFO mapred.LocalJobRunner: 

09/04/07 12:34:36 INFO mapred.TaskRunner: Task:attempt_local_0001_r_000000_0 is done

. And is in the process of commiting

09/04/07 12:34:36 INFO mapred.LocalJobRunner: 

09/04/07 12:34:36 INFO mapred.TaskRunner: Task attempt_local_0001_r_000000_0 is 

allowed to commit now

09/04/07 12:34:36 INFO mapred.FileOutputCommitter: Saved output of task 

'attempt_local_0001_r_000000_0' to file:/Users/tom/workspace/htdg/output

09/04/07 12:34:36 INFO mapred.LocalJobRunner: reduce > reduce

09/04/07 12:34:36 INFO mapred.TaskRunner: Task 'attempt_local_0001_r_000000_0' done.

09/04/07 12:34:36 INFO mapred.JobClient:  map 100% reduce 100%

09/04/07 12:34:36 INFO mapred.JobClient: Job complete: job_local_0001

09/04/07 12:34:36 INFO mapred.JobClient: Counters: 13

09/04/07 12:34:36 INFO mapred.JobClient:   FileSystemCounters

09/04/07 12:34:36 INFO mapred.JobClient:     FILE_BYTES_READ=27571

09/04/07 12:34:36 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=53907

09/04/07 12:34:36 INFO mapred.JobClient:   Map-Reduce Framework

09/04/07 12:34:36 INFO mapred.JobClient:     Reduce input groups=2

09/04/07 12:34:36 INFO mapred.JobClient:     Combine output records=0

09/04/07 12:34:36 INFO mapred.JobClient:     Map input records=5

09/04/07 12:34:36 INFO mapred.JobClient:     Reduce shuffle bytes=0

09/04/07 12:34:36 INFO mapred.JobClient:     Reduce output records=2

09/04/07 12:34:36 INFO mapred.JobClient:     Spilled Records=10

09/04/07 12:34:36 INFO mapred.JobClient:     Map output bytes=45

09/04/07 12:34:36 INFO mapred.JobClient:     Map input bytes=529

09/04/07 12:34:36 INFO mapred.JobClient:     Combine input records=0

09/04/07 12:34:36 INFO mapred.JobClient:     Map output records=5

09/04/07 12:34:36 INFO mapred.JobClient:     Reduce input records=5

% cat output/part-00000

1949	111

1950	22

public class NewMaxTemperature {

  

  static class NewMaxTemperatureMapper

    extends Mapper {



    private static final int MISSING = 9999;

    

    public void map(LongWritable key, Text value, Context context)

        throws IOException, InterruptedException {

      

      String line = value.toString();

      String year = line.substring(15, 19);

      int airTemperature;

      if (line.charAt(87) == '+') { // parseInt doesn't like leading plus signs

        airTemperature = Integer.parseInt(line.substring(88, 92));

      } else {

        airTemperature = Integer.parseInt(line.substring(87, 92));

      }

      String quality = line.substring(92, 93);

      if (airTemperature != MISSING && quality.matches("[01459]")) {

        context.write(new Text(year), new IntWritable(airTemperature));

      }

    }

  }

  

  static class NewMaxTemperatureReducer

    extends Reducer {

  

    public void reduce(Text key, Iterable values,

        Context context)

        throws IOException, InterruptedException {

      

      int maxValue = Integer.MIN_VALUE;

      for (IntWritable value : values) {

        maxValue = Math.max(maxValue, value.get());

      }

      context.write(key, new IntWritable(maxValue));

    }

  }



  public static void main(String[] args) throws Exception {

    if (args.length != 2) {

      System.err.println("Usage: NewMaxTemperature  
");

      System.exit(-1);

    }

    

    Job job = new Job();

    job.setJarByClass(NewMaxTemperature.class);



    FileInputFormat.addInputPath(job, new Path(args[0]));

    FileOutputFormat.setOutputPath(job, new Path(args[1]));

    

    job.setMapperClass(NewMaxTemperatureMapper.class);

    job.setReducerClass(NewMaxTemperatureReducer.class);



    job.setOutputKeyClass(Text.class);

    job.setOutputValueClass(IntWritable.class);

    

    System.exit(job.waitForCompletion(true) ? 0 : 1);

  }

}