The driver parses the four command line arguments to configure this job. It sets our custom input format and calls the static methods to configure it further. All the output is written to the given output directory. The identity mapper is used for this job, and the reduce phase is disabled by setting the number of reduce tasks to zero.

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();

    int numMapTasks = Integer.parseInt(args[0]);
    int numRecordsPerTask = Integer.parseInt(args[1]);
    Path wordList = new Path(args[2]);
    Path outputDir = new Path(args[3]);

    Job job = new Job(conf, "RandomDataGenerationDriver");
    job.setJarByClass(RandomDataGenerationDriver.class);

    job.setNumReduceTasks(0);

    job.setInputFormatClass(RandomStackOverflowInputFormat.class);

    RandomStackOverflowInputFormat.setNumMapTasks(job, numMapTasks);
    RandomStackOverflowInputFormat.setNumRecordPerTask(job,
            numRecordsPerTask);
    RandomStackOverflowInputFormat.setRandomWordList(job, wordList);

    TextOutputFormat.setOutputPath(job, outputDir);

    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(NullWritable.class);

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

The FakeInputSplit class simply extends InputSplit and implements Writable. There is no implementation for any of the overridden methods, or for methods requiring return values return basic values. This input split is used to trick the framework into assigning a task to generate the random data.

public static class FakeInputSplit extends InputSplit implements
        Writable {

    public void readFields(DataInput arg0) throws IOException {
    }

    public void write(DataOutput arg0) throws IOException {
    }

    public long getLength() throws IOException, InterruptedException {
        return 0;
    }

    public String[] getLocations() throws IOException,
            InterruptedException {
        return new String[0];
    }
}

The input format has two main purposes: returning the list of input splits for the framework to generate map tasks from, and then creating the RandomStackOverflowRecordReader for the map task. We override the getSplits method to return a configured number of FakeInputSplit splits. This number is pulled from the configuration. When the framework calls createRecordReader, a RandomStackOverflowRecordReader is instantiated, initialized, and returned.

public static class RandomStackOverflowInputFormat extends
        InputFormat<Text, NullWritable> {

    public static final String NUM_MAP_TASKS = "random.generator.map.tasks";
    public static final String NUM_RECORDS_PER_TASK =
            "random.generator.num.records.per.map.task";
    public static final String RANDOM_WORD_LIST =
            "random.generator.random.word.file";

    public List<InputSplit> getSplits(JobContext job) throws IOException {

        // Get the number of map tasks configured for
        int numSplits = job.getConfiguration().getInt(NUM_MAP_TASKS, -1);

        // Create a number of input splits equivalent to the number of tasks
        ArrayList<InputSplit> splits = new ArrayList<InputSplit>();
        for (int i = 0; i < numSplits; ++i) {
            splits.add(new FakeInputSplit());
        }

        return splits;
    }

    public RecordReader<Text, NullWritable> createRecordReader(
            InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {
        // Create a new RandomStackOverflowRecordReader and initialize it
        RandomStackOverflowRecordReader rr =
                new RandomStackOverflowRecordReader();
        rr.initialize(split, context);
        return rr;
    }

    public static void setNumMapTasks(Job job, int i) {
        job.getConfiguration().setInt(NUM_MAP_TASKS, i);
    }

    public static void setNumRecordPerTask(Job job, int i) {
        job.getConfiguration().setInt(NUM_RECORDS_PER_TASK, i);
    }

    public static void setRandomWordList(Job job, Path file) {
        DistributedCache.addCacheFile(file.toUri(), job.getConfiguration());
    }
}

This record reader is where the data is actually generated. It is given during our FakeInputSplit during initialization, but simply ignores it. The number of records to create is pulled from the job configuration, and the list of random words is read from the DistributedCache. For each call to nextKeyValue, a random record is created using a simple random number generator. The body of the comment is generated by a helper function that randomly selects words from the list, between one and thirty words (also random). The counter is incremented to keep track of how many records have been generated. Once all the records are generated, the record reader returns false, signaling the framework that there is no more input for the mapper.

public static class RandomStackOverflowRecordReader extends
        RecordReader<Text, NullWritable> {

    private int numRecordsToCreate = 0;
    private int createdRecords = 0;
    private Text key = new Text();
    private NullWritable value = NullWritable.get();
    private Random rndm = new Random();
    private ArrayList<String> randomWords = new ArrayList<String>();

    // This object will format the creation date string into a Date
    // object
    private SimpleDateFormat frmt = new SimpleDateFormat(
            "yyyy-MM-dd'T'HH:mm:ss.SSS");

    public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {

        // Get the number of records to create from the configuration
        this.numRecordsToCreate = context.getConfiguration().getInt(
                NUM_RECORDS_PER_TASK, -1);

        // Get the list of random words from the DistributedCache
        URI[] files = DistributedCache.getCacheFiles(context
                .getConfiguration());

        // Read the list of random words into a list
        BufferedReader rdr = new BufferedReader(new FileReader(
                files[0].toString()));

        String line;
        while ((line = rdr.readLine()) != null) {
            randomWords.add(line);
        }
        rdr.close();
    }

    public boolean nextKeyValue() throws IOException,
            InterruptedException {
        // If we still have records to create
        if (createdRecords < numRecordsToCreate) {
            // Generate random data
            int score = Math.abs(rndm.nextInt()) % 15000;
            int rowId = Math.abs(rndm.nextInt()) % 1000000000;
            int postId = Math.abs(rndm.nextInt()) % 100000000;
            int userId = Math.abs(rndm.nextInt()) % 1000000;
            String creationDate = frmt
                    .format(Math.abs(rndm.nextLong()));

            // Create a string of text from the random words
            String text = getRandomText();

            String randomRecord = "<row Id="" + rowId + "" PostId=""
                    + postId + "" Score="" + score + "" Text=""
                    + text + "" CreationDate="" + creationDate
                    + "" UserId"=" + userId + "" />";

            key.set(randomRecord);
            ++createdRecords;
            return true;
        } else {
            // We are done creating records
            return false;
        }
    }

    private String getRandomText() {
        StringBuilder bldr = new StringBuilder();
        int numWords = Math.abs(rndm.nextInt()) % 30 + 1;

        for (int i = 0; i < numWords; ++i) {
            bldr.append(randomWords.get(Math.abs(rndm.nextInt())
                    % randomWords.size())
                    + " ");
        }
        return bldr.toString();
    }

    public Text getCurrentKey() throws IOException,
            InterruptedException {
        return key;
    }

    public NullWritable getCurrentValue() throws IOException,
            InterruptedException {
        return value;
    }

    public float getProgress() throws IOException, InterruptedException {
        return (float) createdRecords / (float) numRecordsToCreate;
    }

    public void close() throws IOException {
        // nothing to do here...
    }
}

The RedisHashOutputFormat is responsible for establishing and verifying the job configuration prior to being submitted to the JobTracker. Once the job has been submitted, it also creates the RecordWriter to serialize all the output key/value pairs. Typically, this is a file in HDFS. However, we are not bound to using HDFS, as we will see in the RecordWriter later on.

The output format contains configuration variables that must be set by the driver to ensure it has all the information required to do its job. Here, we have a couple public static methods to take some of the guess work out of what a developer needs to set. This output format takes in a list of Redis instance hosts as a CSV structure and a Redis hash key to write all the output to. In the checkOutputSpecs method, we ensure that both of these parameters are set before we even both launching the job, as it will surely fail without them. This is where you’ll want to verify your configuration!

The getRecordWriter method is used on the back end to create an instance of a RecordWriter for the map or reduce task. Here, we get the configuration variables required by the RedisHashRecordWriter and return a new instance of it. This record writer is a nested class of the RedisHashOutputFormat, which is not required but is more of a convention. The details of this class are in the following section.

The final method of this output format is getOutputCommitter. The output committer is used by the framework to manage any temporary output before committing in case the task fails and needs to be reexecuted. For this implementation, we don’t typically care whether the task fails and needs to be re-executed. As long as the job finishes we are okay. An output committer is required by the framework, but the NullOutputFormat contains an output committer implementation that doesn’t do anything.

public static class RedisHashOutputFormat extends OutputFormat<Text, Text> {

    public static final String REDIS_HOSTS_CONF =
            "mapred.redishashoutputformat.hosts";
    public static final String REDIS_HASH_KEY_CONF =
            "mapred.redishashinputformat.key";

    public static void setRedisHosts(Job job, String hosts) {
        job.getConfiguration().set(REDIS_HOSTS_CONF, hosts);
    }

    public static void setRedisHashKey(Job job, String hashKey) {
        job.getConfiguration().set(REDIS_HASH_KEY_CONF, hashKey);
    }

    public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext job)
            throws IOException, InterruptedException {
        return new RedisHashRecordWriter(job.getConfiguration().get(
                REDIS_HASH_KEY_CONF), job.getConfiguration().get(
                REDIS_HOSTS_CONF));
    }

    public void checkOutputSpecs(JobContext job) throws IOException {
        String hosts = job.getConfiguration().get(REDIS_HOSTS_CONF);
        if (hosts == null || hosts.isEmpty()) {
            throw new IOException(REDIS_HOSTS_CONF
                    + " is not set in configuration.");
        }

        String hashKey = job.getConfiguration().get(
                REDIS_HASH_KEY_CONF);
        if (hashKey == null || hashKey.isEmpty()) {
            throw new IOException(REDIS_HASH_KEY_CONF
                    + " is not set in configuration.");
        }
    }

    public OutputCommitter getOutputCommitter(TaskAttemptContext context)
            throws IOException, InterruptedException {
        return (new NullOutputFormat<Text, Text>()).getOutputCommitter(context);
    }

    public static class RedisHashRecordWriter extends RecordWriter<Text, Text> {
        // code in next section
    }
}

The RedisHashRecordWriter handles connecting to Redis via the Jedis client and writing out the data. Each key/value pair is randomly written to a Redis instance, providing an even distribution of all data across all Redis instances. The constructor stores the hash key to write to and creates a new Jedis instance.

The code then connects to the Jedis instance and maps it to an integer. This map is used in the write method to get the assigned Jedis instance. The hash code is the key is taken modulo the number of configured Redis instances. The key/value pair is then written to the returned Jedis instance to the configured hash. Finally, all Jedis instances are disconnected in the close method.

public static class RedisHashRecordWriter extends RecordWriter<Text, Text> {

    private HashMap<Integer, Jedis> jedisMap = new HashMap<Integer, Jedis>();
    private String hashKey = null;

    public RedisHashRecordWriter(String hashKey, String hosts) {
        this.hashKey = hashKey;

        // Create a connection to Redis for each host
        // Map an integer 0-(numRedisInstances - 1) to the instance
        int i = 0;
        for (String host : hosts.split(",")) {
            Jedis jedis = new Jedis(host);
            jedis.connect();
            jedisMap.put(i, jedis);
            ++i;
        }
    }

    public void write(Text key, Text value) throws IOException,
            InterruptedException {
        // Get the Jedis instance that this key/value pair will be
        // written to
        Jedis j = jedisMap.get(Math.abs(key.hashCode()) % jedisMap.size());

        // Write the key/value pair
        j.hset(hashKey, key.toString(), value.toString());
    }

    public void close(TaskAttemptContext context) throws IOException,
            InterruptedException {
        // For each jedis instance, disconnect it
        for (Jedis jedis : jedisMap.values()) {
            jedis.disconnect();
        }
    }
}

The Mapper instance is very straightforward and looks like any other mapper. The user ID and reputation are retrieved from the record and then output. The output format does all the heavy lifting for us, allowing it to be reused multiple times to write whatever we want to a Redis hash.

public static class RedisOutputMapper extends
        Mapper<Object, Text, Text, Text> {

    private Text outkey = new Text();
    private Text outvalue = new Text();

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

        Map<String, String> parsed = MRDPUtils.transformXmlToMap(value
                .toString());

        String userId = parsed.get("Id");
        String reputation = parsed.get("Reputation");

        // Set our output key and values
        outkey.set(userId);
        outvalue.set(reputation);

        context.write(outkey, outvalue);
    }
}

The driver code parses the command lines and calls our public static methods to set up writing data to Redis. The job is then submitted just like any other.

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();

    Path inputPath = new Path(args[0]);
    String hosts = args[1];
    String hashName = args[2];

    Job job = new Job(conf, "Redis Output");
    job.setJarByClass(RedisOutputDriver.class);

    job.setMapperClass(RedisOutputMapper.class);
    job.setNumReduceTasks(0);

    job.setInputFormatClass(TextInputFormat.class);
    TextInputFormat.setInputPaths(job, inputPath);

    job.setOutputFormatClass(RedisHashOutputFormat.class);
    RedisHashOutputFormat.setRedisHosts(job, hosts);
    RedisHashOutputFormat.setRedisHashKey(job, hashName);

    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);

    int code = job.waitForCompletion(true) ? 0 : 2;

    System.exit(code);
}

The RedisInputSplit represents the data to be processed by an individual Mapper. In this example, we store the Redis instance hostname as the location of the input split, as well as the hash key. The input split implements the Writable interface, so that it is serializable by the framework, and includes a default constructor in order for the framework to create a new instance via reflection. We return the location via the getLocations method, in the hopes that the JobTracker will assign each map task to a TaskTracker that is hosting the data.

public static class RedisHashInputSplit extends InputSplit implements Writable {

    private String location = null;
    private String hashKey = null;

    public RedisHashInputSplit() {
        // Default constructor for reflection
    }

    public RedisHashInputSplit(String redisHost, String hash) {
        this.location = redisHost;
        this.hashKey = hash;
    }

    public String getHashKey() {
        return this.hashKey;
    }

    public void readFields(DataInput in) throws IOException {
        this.location = in.readUTF();
        this.hashKey = in.readUTF();
    }

    public void write(DataOutput out) throws IOException {
        out.writeUTF(location);
        out.writeUTF(hashKey);
    }

    public long getLength() throws IOException, InterruptedException {
        return 0;
    }

    public String[] getLocations() throws IOException, InterruptedException {
        return new String[] { location };
    }
}

The RedisHashInputFormat mirrors that of the RedisHashOutputFormat in many ways. It contains configuration variables to know which Redis instances to connect to and which hash to read from. In the getSplits method, the configuration is verified and a number of RedisHashInputSplits is created based on the number of Redis hosts. This will create one map task for each configured Redis instance. The Redis hostname and hash key are stored in the input split in order to be retrieved later by the RedisHashRecordReader. The createRecordReader method is called by the framework to get a new instance of a record reader. The record reader’s initialize method is called by the framework, so we can just create a new instance and return it. Again by convention, this class contains two nested classes for the record reader and input split implementations.

public static class RedisHashInputFormat extends InputFormat<Text, Text> {

    public static final String REDIS_HOSTS_CONF =
            "mapred.redishashinputformat.hosts";
    public static final String REDIS_HASH_KEY_CONF =
            "mapred.redishashinputformat.key";
    private static final Logger LOG = Logger
            .getLogger(RedisHashInputFormat.class);

    public static void setRedisHosts(Job job, String hosts) {
        job.getConfiguration().set(REDIS_HOSTS_CONF, hosts);
    }

    public static void setRedisHashKey(Job job, String hashKey) {
        job.getConfiguration().set(REDIS_HASH_KEY_CONF, hashKey);
    }

    public List<InputSplit> getSplits(JobContext job) throws IOException {
        String hosts = job.getConfiguration().get(REDIS_HOSTS_CONF);

        if (hosts == null || hosts.isEmpty()) {
            throw new IOException(REDIS_HOSTS_CONF
                    + " is not set in configuration.");
        }

        String hashKey = job.getConfiguration().get(REDIS_HASH_KEY_CONF);
        if (hashKey == null || hashKey.isEmpty()) {
            throw new IOException(REDIS_HASH_KEY_CONF
                    + " is not set in configuration.");
        }

        // Create an input split for each host
        List<InputSplit> splits = new ArrayList<InputSplit>();
        for (String host : hosts.split(",")) {
            splits.add(new RedisHashInputSplit(host, hashKey));
        }

        LOG.info("Input splits to process: " + splits.size());
        return splits;
    }

    public RecordReader<Text, Text> createRecordReader(InputSplit split,
            TaskAttemptContext context) throws IOException,
            InterruptedException {
        return new RedisHashRecordReader();
    }

    public static class RedisHashRecordReader extends RecordReader<Text, Text> {
        // code in next section
    }

    public static class RedisHashInputSplit extends 
            InputSplit implements Writable {
        // code in next section
    }
}

The RedisHashRecordReader is where most of the work is done. The initialize method is called by the framework and provided with an input split we created in the input format. Here, we get the Redis instance to connect to and the hash key. We then connect to Redis and get the number of key/value pairs we will be reading from Redis. The hash doesn’t have a means to iterate or stream the data one at a time or in bulk, so we simply pull everything over and disconnect from Redis. We store an iterator over the entries and log some helpful statements along the way.

In nextKeyValue, we iterate through the map of entries one at a time and set the record reader’s writable objects for the key and value. A return value of true informs the framework that there is a key/value pair to process. Once we have exhausted all the key/value pairs, false is returned so the map task can complete.

The other methods of the record reader are used by the framework to get the current key and value for the mapper to process. It is worthwhile to reuse this object whenever possible. The getProgress method is useful for reporting gradual status to the JobTracker and should also be reused if possible. Finally, the close method is for finalizing the process. Since we pulled all the information and disconnected from Redis in the initialize method, there is nothing to do here.

public static class RedisHashRecordReader extends RecordReader<Text, Text> {

    private static final Logger LOG =
            Logger.getLogger(RedisHashRecordReader.class);
    private Iterator<Entry<String, String>> keyValueMapIter = null;
    private Text key = new Text(), value = new Text();
    private float processedKVs = 0, totalKVs = 0;
    private Entry<String, String> currentEntry = null;

    public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {
        // Get the host location from the InputSplit
        String host = split.getLocations()[0];
        String hashKey = ((RedisHashInputSplit) split).getHashKey();

        LOG.info("Connecting to " + host + " and reading from "
                + hashKey);

        Jedis jedis = new Jedis(host);
        jedis.connect();
        jedis.getClient().setTimeoutInfinite();

        // Get all the key/value pairs from the Redis instance and store
        // them in memory
        totalKVs = jedis.hlen(hashKey);
        keyValueMapIter = jedis.hgetAll(hashKey).entrySet().iterator();
        LOG.info("Got " + totalKVs + " from " + hashKey);
        jedis.disconnect();
    }

    public boolean nextKeyValue() throws IOException,
            InterruptedException {

        // If the key/value map still has values
        if (keyValueMapIter.hasNext()) {
            // Get the current entry and set the Text objects to the entry
            currentEntry = keyValueMapIter.next();
            key.set(currentEntry.getKey());
            value.set(currentEntry.getValue());
            return true;
        } else {
            // No more values? return false.
            return false;
        }
    }

    public Text getCurrentKey() throws IOException,
            InterruptedException {
        return key;
    }

    public Text getCurrentValue() throws IOException,
            InterruptedException {
        return value;
    }

    public float getProgress() throws IOException, InterruptedException {
        return processedKVs / totalKVs;
    }

    public void close() throws IOException {
        // nothing to do here
    }
}

Much like the previous example’s driver, we use the public static methods provided by the input format to modify the job configuration. Since we are just using the identity mapper, we don’t need to set any special classes. The number of reduce tasks is set to zero to specify that this is a map-only job.

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();

    String hosts = otherArgs[0];
    String hashKey = otherArgs[1];
    Path outputDir = new Path(otherArgs[2]);

    Job job = new Job(conf, "Redis Input");
    job.setJarByClass(RedisInputDriver.class);

    // Use the identity mapper
    job.setNumReduceTasks(0);

    job.setInputFormatClass(RedisHashInputFormat.class);
    RedisHashInputFormat.setRedisHosts(job, hosts);
    RedisHashInputFormat.setRedisHashKey(job, hashKey);

    job.setOutputFormatClass(TextOutputFormat.class);
    TextOutputFormat.setOutputPath(job, outputDir);

    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);

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

To help better store information, a custom WritableComparable is implemented in order to allow the mapper to set information needed by the record writer. This class contains methods to set and get the field name to be stored in Redis, as well as the last access month. The last access month accepts a zero-based integer value for the month, but is later turned into a string representation for easier querying in the next example. Take the time to implement the compareTo, toString, and hashCode methods (like every good Java developer!).

public static class RedisKey implements WritableComparable<RedisKey> {

    private int lastAccessMonth = 0;
    private Text field = new Text();

    public int getLastAccessMonth() {
        return this.lastAccessMonth;
    }

    public void setLastAccessMonth(int lastAccessMonth) {
        this.lastAccessMonth = lastAccessMonth;
    }

    public Text getField() {
        return this.field;
    }

    public void setField(String field) {
        this.field.set(field);
    }

    public void readFields(DataInput in) throws IOException {
        lastAccessMonth = in.readInt();
        this.field.readFields(in);
    }

    public void write(DataOutput out) throws IOException {
        out.writeInt(lastAccessMonth);
        this.field.write(out);
    }

    public int compareTo(RedisKey rhs) {
        if (this.lastAccessMonth == rhs.getLastAccessMonth()) {
            return this.field.compareTo(rhs.getField());
        } else {
            return this.lastAccessMonth < rhs.getLastAccessMonth() ? -1 : 1;
        }
    }

    public String toString() {
        return this.lastAccessMonth + "	" + this.field.toString();
    }

    public int hashCode() {
        return toString().hashCode();
    }
}

This output format is extremely basic, as all the grunt work is handled in the record writer. The main thing to focus on is the templated arguments when extending the OutputFormat class. This output format accepts our custom class as the output key and a Text object as the output value. Any other classes will cause errors when trying to write any output.

Since our record writer implementation is coded to a specific and known output, there is no need to verify any output specification of the job. An output committer is still required by the framework, so we use NullOutputFormat’s output committer.

public static class RedisLastAccessOutputFormat
        extends OutputFormat<RedisKey, Text> {

    public RecordWriter<RedisKey, Text> getRecordWriter(
            TaskAttemptContext job) throws IOException, InterruptedException {
        return new RedisLastAccessRecordWriter();
    }

    public void checkOutputSpecs(JobContext context) throws IOException,
            InterruptedException {
    }

    public OutputCommitter getOutputCommitter(TaskAttemptContext context)
            throws IOException, InterruptedException {
        return (new NullOutputFormat<Text, Text>()).getOutputCommitter(context);
    }

    public static class RedisLastAccessRecordWriter
            extends RecordWriter<RedisKey, Text> {
        // Code in next section
    }
}

The RedisLastAccessRecordWriter is templated to accept the same classes as the output format. The construction of the class connects to all six Redis instances and puts them in a map. This map stores the month-to-Redis-instance mappings and is used in the write method to retrieve the proper instance. The write method then uses a map of month int to a three character month code for serialization. This map is omitted for brevity, but looks something like 0→JAN, 1→FEB, ..., 11→DEC. This means all the hashes in Redis are named based on the three-character month code. The close method disconnects all the Redis instances.

public static class RedisLastAccessRecordWriter
        extends RecordWriter<RedisKey, Text> {

    private HashMap<Integer, Jedis> jedisMap = new HashMap<Integer, Jedis>();

    public RedisLastAccessRecordWriter() {
        // Create a connection to Redis for each host
        int i = 0;
        for (String host : MRDPUtils.REDIS_INSTANCES) {
            Jedis jedis = new Jedis(host);
            jedis.connect();
            jedisMap.put(i, jedis);
            jedisMap.put(i + 1, jedis);
            i += 2;
        }
    }

    public void write(RedisKey key, Text value) throws IOException,
            InterruptedException {
        // Get the Jedis instance that this key/value pair will be
        // written to -- (0,1)->0, (2-3)->1, ... , (10-11)->5
        Jedis j = jedisMap.get(key.getLastAccessMonth());

        // Write the key/value pair
        j.hset(MONTH_FROM_INT.get(key.getLastAccessMonth()), key
                .getField().toString(), value.toString());
    }

    public void close(TaskAttemptContext context) throws IOException,
            InterruptedException {
        // For each jedis instance, disconnect it
        for (Jedis jedis : jedisMap.values()) {
            jedis.disconnect();
        }
    }
}

The mapper code parses each input record and sets the values for the output RedisKey and the output value. The month of the last access data is parsed via the Calendar and SimpleDateFormat classes.

public static class RedisLastAccessOutputMapper extends
        Mapper<Object, Text, RedisKey, Text> {

    // This object will format the creation date string into a Date object
    private final static SimpleDateFormat frmt = new SimpleDateFormat(
            "yyyy-MM-dd'T'HH:mm:ss.SSS");

    private RedisKey outkey = new RedisKey();
    private Text outvalue = new Text();

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

        Map<String, String> parsed = MRDPUtils.transformXmlToMap(value
                .toString());

        String userId = parsed.get("Id");
        String reputation = parsed.get("Reputation");

        // Grab the last access date
        String strDate = parsed.get("LastAccessDate");

        // Parse the string into a Calendar object
        Calendar cal = Calendar.getInstance();
        cal.setTime(frmt.parse(strDate));

        // Set our output key and values
        outkey.setLastAccessMonth(cal.get(Calendar.MONTH));
        outkey.setField(userId);
        outvalue.set(reputation);

        context.write(outkey, outvalue);
    }
}

The driver looks very similar to a more basic job configuration. All the special configuration is entirely handled by the output format class and record writer.

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();

    Path inputPath = new Path(args[0]);

    Job job = new Job(conf, "Redis Last Access Output");
    job.setJarByClass(PartitionPruningOutputDriver.class);

    job.setMapperClass(RedisLastAccessOutputMapper.class);
    job.setNumReduceTasks(0);

    job.setInputFormatClass(TextInputFormat.class);
    TextInputFormat.setInputPaths(job, inputPath);

    job.setOutputFormatClass(RedisHashSetOutputFormat.class);

    job.setOutputKeyClass(RedisKey.class);
    job.setOutputValueClass(Text.class);

    int code = job.waitForCompletion(true) ? 0 : 2;

    System.exit(code);
}

The input split shown here is very similar to the input split in External Source Input Example. Instead of storing one hash key, we are going to store multiple hash keys. This is because the data is partitioned based on month, instead of all the data being randomly distributed in one hash.

public static class RedisLastAccessInputSplit
        extends InputSplit implements Writable {

    private String location = null;
    private List<String> hashKeys = new ArrayList<String>();

    public RedisLastAccessInputSplit() {
        // Default constructor for reflection
    }

    public RedisLastAccessInputSplit(String redisHost) {
        this.location = redisHost;
    }

    public void addHashKey(String key) {
        hashKeys.add(key);
    }

    public void removeHashKey(String key) {
        hashKeys.remove(key);
    }

    public List<String> getHashKeys() {
        return hashKeys;
    }

    public void readFields(DataInput in) throws IOException {
        location = in.readUTF();
        int numKeys = in.readInt();
        hashKeys.clear();
        for (int i = 0; i < numKeys; ++i) {
            hashKeys.add(in.readUTF());
        }
    }

    public void write(DataOutput out) throws IOException {
        out.writeUTF(location);
        out.writeInt(hashKeys.size());
        for (String key : hashKeys) {
            out.writeUTF(key);
        }
    }

    public long getLength() throws IOException, InterruptedException {
        return 0;
    }

    public String[] getLocations() throws IOException, InterruptedException {
        return new String[] { location };
    }
}

This input format class intelligently creates RedisLastAccessInputSplit objects from the selected months of data. Much like the output format we showed earlier in OutputFormat code, this output format writes RedisKey objects, this input format reads the same objects and is templated to enforce this on mapper implementations. It initially creates a hash map of host-to-input splits in order to add the hash keys to the input split, rather than adding both months of data to the same split. If a split has not been created for a particular month, a new one is created and the month hash key is added. Otherwise, the hash key is added to the split that has already been created. A List is then created out of the values stored in the map. This will create a number of input splits equivalent to the number of Redis instances required to satisfy the query.

There are a number of helpful hash maps to help convert a month string to an integer, as well as figure out which Redis instance hosts which month of data. The initialization of these hash maps are ommitted from the static block for brevity.

public static class RedisLastAccessInputFormat
        extends InputFormat<RedisKey, Text> {

    public static final String REDIS_SELECTED_MONTHS_CONF =
            "mapred.redilastaccessinputformat.months";
    private static final HashMap<String, Integer> MONTH_FROM_STRING =
            new HashMap<String, Integer>();
    private static final HashMap<String, String> MONTH_TO_INST_MAP =
            new HashMap<String, String>();
    private static final Logger LOG = Logger
            .getLogger(RedisLastAccessInputFormat.class);

    static {
        // Initialize month to Redis instance map
        // Initialize month 3 character code to integer
    }

    public static void setRedisLastAccessMonths(Job job, String months) {
        job.getConfiguration().set(REDIS_SELECTED_MONTHS_CONF, months);
    }

    public List<InputSplit> getSplits(JobContext job) throws IOException {

        String months = job.getConfiguration().get(
                REDIS_SELECTED_MONTHS_CONF);

        if (months == null || months.isEmpty()) {
            throw new IOException(REDIS_SELECTED_MONTHS_CONF
                    + " is null or empty.");
        }

        // Create input splits from the input months
        HashMap<String, RedisLastAccessInputSplit> instanceToSplitMap =
                    new HashMap<String, RedisLastAccessInputSplit>();

        for (String month : months.split(",")) {
            String host = MONTH_TO_INST_MAP.get(month);
            RedisLastAccessInputSplit split = instanceToSplitMap.get(host);
            if (split == null) {
                split = new RedisLastAccessInputSplit(host);
                split.addHashKey(month);
                instanceToSplitMap.put(host, split);
            } else {
                split.addHashKey(month);
            }
        }

        LOG.info("Input splits to process: " +
                instanceToSplitMap.values().size());
        return new ArrayList<InputSplit>(instanceToSplitMap.values());
    }

    public RecordReader<RedisKey, Text> createRecordReader(
            InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {
        return new RedisLastAccessRecordReader();
    }

    public static class RedisLastAccessRecordReader
            extends RecordReader<RedisKey, Text> {
            // Code in next section
    }
}

The RedisLastAccessRecordReader is a bit more complicated than the other record readers we have seen. It needs to read from multiple hashes, rather than just reading everything at once in the initialize method. Here, the configuration is simply read in this method.

In nextKeyValue, a new connection to Redis is created if the iterator through the hash is null, or if we have reached the end of all the hashes to read. If the iterator through the hashes does not have a next value, we immediately return false, as there is no more data for the map task. Otherwise, we connect to Redis and pull all the data from the specific hash. The hash iterator is then used to exhaust all the field value pairs from Redis. A do-while loop is used to ensure that once a hash iterator is complete, it will loop back around to get data from the next hash or inform the task there is no more data to be read.

The implementation of the remaining methods are identical to that of the RedisHashRecordReader and are omitted.

public static class RedisLastAccessRecordReader
        extends RecordReader<RedisKey, Text> {

    private static final Logger LOG = Logger
            .getLogger(RedisLastAccessRecordReader.class);
    private Entry<String, String> currentEntry = null;
    private float processedKVs = 0, totalKVs = 0;
    private int currentHashMonth = 0;
    private Iterator<Entry<String, String>> hashIterator = null;
    private Iterator<String> hashKeys = null;
    private RedisKey key = new RedisKey();
    private String host = null;
    private Text value = new Text();

    public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {

        // Get the host location from the InputSplit
        host = split.getLocations()[0];

        // Get an iterator of all the hash keys we want to read
        hashKeys = ((RedisLastAccessInputSplit) split)
                .getHashKeys().iterator();

        LOG.info("Connecting to " + host);
    }
    public boolean nextKeyValue() throws IOException,
            InterruptedException {

        boolean nextHashKey = false;
        do {
            // if this is the first call or the iterator does not have a
            // next
            if (hashIterator == null || !hashIterator.hasNext()) {
                // if we have reached the end of our hash keys, return
                // false
                if (!hashKeys.hasNext()) {
                    // ultimate end condition, return false
                    return false;
                } else {
                    // Otherwise, connect to Redis and get all
                    // the name/value pairs for this hash key
                    Jedis jedis = new Jedis(host);
                    jedis.connect();
                    String strKey = hashKeys.next();
                    currentHashMonth = MONTH_FROM_STRING.get(strKey);
                    hashIterator = jedis.hgetAll(strKey).entrySet()
                            .iterator();
                    jedis.disconnect();
                }
            }

            // If the key/value map still has values
            if (hashIterator.hasNext()) {
                // Get the current entry and set 
                // the Text objects to the entry
                currentEntry = hashIterator.next();
                key.setLastAccessMonth(currentHashMonth);
                key.setField(currentEntry.getKey());
                value.set(currentEntry.getValue());
            } else {
                nextHashKey = true;
            }
        } while (nextHashKey);

        return true;
    }

    ...
}

The driver code sets the months most recently accessed passed in via the command line. This configuration parameter is used by the input format to determine which Redis instances to read from, rather than reading from every Redis instance. It also sets the output directory for the job. Again, it uses the identity mapper rather than performing any analysis on the data retrieved.

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();

    String lastAccessMonths = args[0];
    Path outputDir = new Path(args[1]);

    Job job = new Job(conf, "Redis Input");
    job.setJarByClass(PartitionPruningInputDriver.class);

    // Use the identity mapper
    job.setNumReduceTasks(0);

    job.setInputFormatClass(RedisLastAccessInputFormat.class);
    RedisLastAccessInputFormat.setRedisLastAccessMonths(job,
            lastAccessMonths);

    job.setOutputFormatClass(TextOutputFormat.class);
    TextOutputFormat.setOutputPath(job, outputDir);

    job.setOutputKeyClass(RedisKey.class);
    job.setOutputValueClass(Text.class);

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