Hadoop MapReduce Features : Custom Data Types

Hadoop requires every data type to be used as keys to implement Writable and Comparable interfaces and every data type to be used as values to implement Writable interface.

Writable interface

Writable interface provides a way to serialize and deserialize data, across network. Located in org.apache.hadoop.io package.

public interface Writable {
   void write(DataOutput out) throws IOException;
   void readFields(DataInput in) throws IOException;
}

Comparable interface

Hadoop uses Java's Comparable interface to facilitate the sorting process. Located in java.lang package.

public interface Comparable<T> {
   public int compareTo(T o);
}

WritableComparable interface

For convenience Hadoop provides a WritableComparable interface which wraps both Writable and Comparable interfaces to a single interface. Located in org.apache.hadoop.io package.

public interface WritableComparable<T> extends Writable, Comparable<T> {
}

Hadoop provides a set of classes; Text, IntWritable, LongWritable, FloatWritable, BooleanWritable etc..., which implement WritableComparable interface, and therefore can be straightly used as key and value types.

Custom Data Types

Example

Assume you want an object representation of a vehicle to be the type of your key or value. Three properties of a vehicle has taken in to consideration.

1. Manufacturer
2. Vehicle Identification Number(VIN)
3. Mileage

Note that the Vehicle Identification Number(VIN) is unique for each vehicle.

A tab delimited file containing the observations of these three variables contains following sample data in it.

Sample Data

Toyota 1GCCS148X48370053 10000
Toyota 1FAPP64R1LH452315 40000
BMW WP1AA29P58L263510 10000
BMW JM3ER293470820653 60000
Nissan 3GTEC14V57G579789 10000
Nissan 1GNEK13T6YJ290558 25000
Honda 1GC4KVBG6AF244219 10000
Honda 1FMCU5K39AK063750 30000

Custom Value Types

Hadoop provides the freedom of creating custom value types by implementing the Writable interface. Implementing the writable interface one should implement its two abstract methods write() and readFields(). In addition to that, in the following example java code, I have overridden the toString() method to return the text representation of the object.

package net.eviac.blog.datatypes.value;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Writable;

/**
 * @author pavithra
 * 
 * Custom data type to be used as a value in Hadoop.
 * In Hadoop every data type to be used as values must implement Writable interface.
 *
 */
public class Vehicle implements Writable {

  private String model;
  private String vin;
  private int mileage;

  public void write(DataOutput out) throws IOException {
    out.writeUTF(model);
    out.writeUTF(vin);
    out.writeInt(mileage);
  }

  public void readFields(DataInput in) throws IOException {
    model = in.readUTF();
    vin = in.readUTF();
    mileage = in.readInt();
  }

  @Override
  public String toString() {
    return model + ", " + vin + ", "
        + Integer.toString(mileage);
  }

  public String getModel() {
    return model;
  }
  public void setModel(String model) {
    this.model = model;
  }
  public String getVin() {
    return vin;
  }
  public void setVin(String vin) {
    this.vin = vin;
  }
  public int getMileage() {
    return mileage;
  }
  public void setMileage(int mileage) {
    this.mileage = mileage;
  }
  
}

Following MapReduce job, outputs total Mileage per Manufacturer, with using Vehicle as a custom value type.

package net.eviac.blog.datatypes.jobs;

import java.io.IOException;

import net.eviac.blog.datatypes.value.Vehicle;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;
import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.Logger;

/**
 * @author pavithra
 *
 */
public class ModelTotalMileage {
  
  static final Logger logger = Logger.getLogger(ModelTotalMileage.class);

  public static class ModelMileageMapper
  extends Mapper<Object, Text, Text, Vehicle>{

    private Vehicle vehicle = new Vehicle();
    private Text model = new Text();

    public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {

      String var[] = new String[6];
      var = value.toString().split("\t"); 

      if(var.length == 3){
        model.set(var[0]);
        vehicle.setModel(var[0]);
        vehicle.setVin(var[1]);
        vehicle.setMileage(Integer.parseInt(var[2]));
        context.write(model, vehicle);
      }
    }
  }

  public static class ModelTotalMileageReducer
  extends Reducer<Text,Vehicle,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<Vehicle> values,
        Context context
        ) throws IOException, InterruptedException {
      int totalMileage = 0;
      for (Vehicle vehicle : values) {
        totalMileage += vehicle.getMileage();
      }
      result.set(totalMileage);
      context.write(key, result);
    }
  }

  public static void main(String[] args) throws Exception {
    BasicConfigurator.configure();
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "Model Total Mileage");
    job.setJarByClass(ModelTotalMileage.class);
    job.setMapperClass(ModelMileageMapper.class);
    job.setReducerClass(ModelTotalMileageReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Vehicle.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }

}

Output

BMW 70000
Honda 40000
Nissan 35000
Toyota 50000

Custom Key Types

Hadoop provides the freedom of creating custom key types by implementing the WritableComparable interface. In addition to using Writable interface, so they can be transmitted over the network, keys must implement Java's Comparable interface to facilitate the sorting process. Since outputs are sorted on keys by the framework, only the data type to be used as keys must implement Comparable interface.

Implementing Comparable interface a class should implement its abstract method compareTo(). A Vehicle object must be able to be compared to other vehicle objects, to facilitate the sorting process. Sorting process uses compareTo(), to determine how Vehicle objects should be sorted. For an instance, since VIN is a String and String implements Comparable, we can sort vehicles by VIN.

For partitioning process, it is important for key types to implement hashCode() as well, thus should override the equals() as well. hashCode() should use the same variable as equals() which in our case is VIN. If equals() method says two Vehicle objects are equal if they have the same VIN, Vehicle objects with the same VIN will have to return identical hash codes.

Following example provides a java code for complete custom key type.

package net.eviac.blog.datatypes.key;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

/**
 * @author pavithra
 * 
 * Custom data type to be used as a key in Hadoop.
 * In Hadoop every data type to be used as keys must implement WritableComparable interface.
 *
 */
public class Vehicle implements WritableComparable<Vehicle> {

  private String model;
  private String vin;
  private int mileage;

  public void write(DataOutput out) throws IOException {
    out.writeUTF(model);
    out.writeUTF(vin);
    out.writeInt(mileage);
  }

  public void readFields(DataInput in) throws IOException {
    model = in.readUTF();
    vin = in.readUTF();
    mileage = in.readInt();
  }

  @Override
  public String toString() {
    return model + ", " + vin + ", "
        + Integer.toString(mileage);
  }
  
  public int compareTo(Vehicle o) {
    return vin.compareTo(o.getVin());
  } 
  
  @Override
  public boolean equals(Object obj) {
    if((obj instanceof Vehicle) && (((Vehicle)obj).getVin().equals(vin))){
      return true;
    }else {
      return false;
    }    
  }
  
  @Override
  public int hashCode() {
    int ascii = 0;
    for(int i=1;i<=vin.length();i++){
      char character = vin.charAt(i);
      ascii += (int)character;
    }
    return ascii;
  }

  public String getModel() {
    return model;
  }
  public void setModel(String model) {
    this.model = model;
  }
  public String getVin() {
    return vin;
  }
  public void setVin(String vin) {
    this.vin = vin;
  }
  public int getMileage() {
    return mileage;
  }
  public void setMileage(int mileage) {
    this.mileage = mileage;
  }  
  
}

Following MapReduce job, outputs Mileage per vehicle, with using Vehicle as a custom key type.

package net.eviac.blog.datatypes.jobs;

import java.io.IOException;

import net.eviac.blog.datatypes.key.Vehicle;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;
import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.Logger;

/**
 * @author pavithra
 *
 */
public class VehicleMileage {
  
  static final Logger logger = Logger.getLogger(VehicleMileage.class);

  public static class VehicleMileageMapper
  extends Mapper<Object, Text, Vehicle, IntWritable>{

    private Vehicle vehicle = new Vehicle();
    private IntWritable mileage = new IntWritable();

    public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {

      String var[] = new String[6];
      var = value.toString().split("\t"); 

      if(var.length == 3){
        mileage.set(Integer.parseInt(var[2]));
        vehicle.setModel(var[0]);
        vehicle.setVin(var[1]);
        vehicle.setMileage(Integer.parseInt(var[2]));
        context.write(vehicle, mileage);
      }
    }
  }

  public static class VehicleMileageReducer
  extends Reducer<Vehicle,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();
    private Text vin = new Text();

    public void reduce(Vehicle key, Iterable<IntWritable> values,
        Context context
        ) throws IOException, InterruptedException {
      int totalMileage = 0;
      for (IntWritable mileage : values) {
        totalMileage += mileage.get();
      }
      result.set(totalMileage);
      vin.set(key.getVin());
      context.write(vin, result);
    }
  }

  public static void main(String[] args) throws Exception {
    BasicConfigurator.configure();
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "Model Total Mileage");
    job.setJarByClass(VehicleMileage.class);
    job.setMapperClass(VehicleMileageMapper.class);
    job.setReducerClass(VehicleMileageReducer.class);
    job.setMapOutputKeyClass(Vehicle.class);
    job.setOutputKeyClass(Text.class);    
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }

}

Output

1FAPP64R1LH452315 40000
1FMCU5K39AK063750 30000
1GC4KVBG6AF244219 10000
1GCCS148X48370053 10000
1GNEK13T6YJ290558 25000
3GTEC14V57G579789 10000
JM3ER293470820653 60000
WP1AA29P58L263510 10000