MapReduce uses a Java class InputFormat to read input data. Most frequently, these classes read data directly from HDFS. InputFormat implementations also exist to read data from HBase, Cassandra, and other data sources. The task of the InputFormat is twofold. First, it determines how data is split into sections so that it can be processed in parallel by MapReduce’s map tasks. Second, it provides a RecordReader, a class that MapReduce uses to read records from its input source and convert them to keys and values for the map task to operate on.

HCatalog provides HCatInputFormat to enable MapReduce users to read data stored in Hive’s data warehouse. It allows users to read only the partitions of tables and columns that they need. And it provides the records in a convenient list format so that users do not need to parse them.

When initializing HCatInputFormat, the first thing to do is specify the table to be read. This is done by creating an InputJobInfo class and specifying the database, table, and partition filter to use.

InputJobInfo.java

  /**
   * Initializes a new InputJobInfo
   * for reading data from a table.
   * @param databaseName the db name
   * @param tableName the table name
   * @param filter the partition filter
   */
  public static InputJobInfo create(String databaseName,
      String tableName,
      String filter) {
      ...
  }

databaseName name indicates the Hive database (or schema) the table is in. If this is null then the default database will be used. The tableName is the table that will be read. This must be non-null and refer to a valid table in Hive. filter indicates which partitions the user wishes to read. If it is left null then the entire table will be read. Care should be used here, as reading all the partitions of a large table can result in scanning a large volume of data.

Filters are specified as an SQL-like where clause. They should reference only partition columns of the data. For example, if the table to be read is partitioned on a column called datestamp, the filter might look like datestamp = "2012-05-26". Filters can contain =, >, >=, <, <=, and, and or as operators.

There is a bug in the ORM mapping layer used by Hive v0.9.0 and earlier that causes filter clauses with >, >=, <, or <= to fail.

This InputJobInfo instance is then passed to HCatInputFormat via the method setInput along with the instance of Job being used to configure the MapReduce job:

Job job = new Job(conf, "Example");
InputJobInfo inputInfo =  InputJobInfo.create(dbName, inputTableName, filter));
HCatInputFormat.setInput(job, inputInfo);

The map task will need to specify HCatRecord as a value type. The key type is not important, as HCatalog does not provide keys to the map task. For example, a map task that reads data via HCatalog might look like:

  public static class Map extends
      Mapper<WritableComparable, HCatRecord, Text, Text> {

    @Override
    protected void map(
      WritableComparable key,
      HCatRecord value,
      org.apache.hadoop.mapreduce.Mapper<WritableComparable,
          HCatRecord, Text, HCatRecord>.Context context) {
        ...
    }
  }

HCatRecord is the class that HCatalog provides for interacting with records. It presents a simple get and set interface. Records can be requested by position or by name. When requesting columns by name, the schema must be provided, as each individual HCatRecord does not keep a reference to the schema. The schema can be obtained by calling HCatInputFormat.getOutputSchema(). Since Java does not support overloading of functions by return type, different instances of get and set are provided for each data type. These methods use the object versions of types rather than scalar versions (that is java.lang.Integer rather than int). This allows them to express null as a value. There are also implementations of get and set that work with Java Objects:

// get the first column, as an Object and cast it to a Long
Long cnt = record.get(0);

// get the column named "cnt" as a Long
Long cnt = record.get("cnt", schema);

// set the column named "user" to the string "fred"
record.setString("user", schema, "fred");

Often a program will not want to read all of the columns in an input. In this case it makes sense to trim out the extra columns as quickly as possible. This is particularly true in columnar formats like RCFile, where trimming columns early means reading less data from disk. This can be achieved by passing a schema that describes the desired columns. This must be done during job configuration time. The following example will configure the user’s job to read only two columns named user and url:

HCatSchema baseSchema = HCatBaseInputFormat.getOutputSchema(context);
List<HCatFieldSchema> fields = new List<HCatFieldSchema>(2);
fields.add(baseSchema.get("user"));
fields.add(baseSchema.get("url"));
HCatBaseInputFormat.setOutputSchema(job, new HCatSchema(fields));

Similar to reading data, when writing data, the database and table to be written to need to be specified. If the data is being written to a partitioned table and only one partition is being written, then the partition to be written needs to be specified as well:

  /**
   * Initializes a new OutputJobInfo instance for writing data from a table.
   * @param databaseName the db name
   * @param tableName the table name
   * @param partitionValues The partition values to publish to, can be null or empty Map
   */
  public static OutputJobInfo create(String databaseName,
                                     String tableName,
                                     Map<String, String> partitionValues) {
      ...
  }

The databaseName name indicates the Hive database (or schema) the table is in. If this is null then the default database will be used. The tableName is the table that will be written to. This must be non-null and refer to a valid table in Hive. partitionValues indicates which partition the user wishes to create. If only one partition is to be written, the map must uniquely identify a partition. For example, if the table is partitioned by two columns, entries for both columns must be in the map. When working with tables that are not partitioned, this field can be left null. When the partition is explicitly specified in this manner, the partition column need not be present in the data. If it is, it will be removed by HCatalog before writing the data to the Hive warehouse, as Hive does not store partition columns with the data.

It is possible to write to more than one partition at a time. This is referred to as dynamic partitioning, because the records are partitioned dynamically at runtime. For dynamic partitioning to be used, the values of the partition column(s) must be present in the data. For example, if a table is partitioned by a column “datestamp,” that column must appear in the data collected in the reducer. This is because HCatalog will read the partition column(s) to determine which partition to write the data to. As part of writing the data, the partition column(s) will be removed.

Once an OutputJobInfo has been created, it is then passed to HCatOutputFormat via the static method setOutput:

OutputJobInfo outputInfo = OutputJobInfo.create(dbName, outputTableName, null));
HCatOutputFormat.setOutput(job, outputInfo);

When writing with HCatOutputFormat, the output key type is not important. The value must be HCatRecord. Records can be written from the reducer, or in map only jobs from the map task.

Putting all this together in an example, the following code will read a partition with a datestamp of 20120531 from the table rawevents, count the number of events for each user, and write the result to a table cntd:

public class MRExample extends Configured implements Tool {

  public static class Map extends
    Mapper<WritableComparable, HCatRecord, Text, LongWritable> {

    protected void map(WritableComparable key,
                       HCatRecord value,
                       Mapper<WritableComparable, HCatRecord,
                         Text, LongWritable>.Context context)
                       throws IOException, InterruptedException {
      // Get our schema from the Job object.
      HCatSchema schema = HCatBaseInputFormat.getOutputSchema(context);

      // Read the user field
      String user = value.get("user", schema);
      context.write(new Text(user), new LongWritable(1));
    }
  }

  public static class Reduce extends Reducer<Text, LongWritable,
      WritableComparable, HCatRecord> {

    protected void reduce(Text key, Iterable<LongWritable> values,
        Reducer<Text, LongWritable,
        WritableComparable, HCatRecord>.Context context)
        throws IOException ,InterruptedException {

      List<HCatFieldSchema> columns = new ArrayList<HCatFieldSchema>(2);
      columns.add(new HCatFieldSchema("user", HCatFieldSchema.Type.STRING, ""));
      columns.add(new HCatFieldSchema("cnt", HCatFieldSchema.Type.BIGINT, ""));
      HCatSchema schema = new HCatSchema(columns);

      long sum = 0;
      Iterator<IntWritable> iter = values.iterator();
      while (iter.hasNext()) sum += iter.next().getLong();
      HCatRecord output = new DefaultHCatRecord(2);
      record.set("user", schema, key.toString());
      record.setLong("cnt", schema, sum);
      context.write(null, record);
    }
  }

  public int run(String[] args) throws Exception {
    Job job = new Job(conf, "Example");
    // Read the "rawevents" table, partition "20120531", in the default
    // database
    HCatInputFormat.setInput(job, InputJobInfo.create(null, "rawevents",
        "datestamp='20120531'"));
    job.setInputFormatClass(HCatInputFormat.class);
    job.setJarByClass(MRExample.class);
    job.setMapperClass(Map.class);
    job.setReducerClass(Reduce.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(LongWritable.class);
    job.setOutputKeyClass(WritableComparable.class);
    job.setOutputValueClass(DefaultHCatRecord.class);
    // Write into "cntd" table, partition "20120531", in the default database
    HCatOutputFormat.setOutput(job
        OutputJobInfo.create(null, "cntd", "ds=20120531"));
    job.setOutputFormatClass(HCatOutputFormat.class);
    return (job.waitForCompletion(true) ? 0 : 1);
    }

    public static void main(String[] args) throws Exception {
      int exitCode = ToolRunner.run(new MRExample(), args);
      System.exit(exitCode);
    }
}