{"id":664,"date":"2009-12-20T11:27:44","date_gmt":"2009-12-20T16:27:44","guid":{"rendered":"http:\/\/philippeadjiman.com\/blog\/?p=664"},"modified":"2025-07-18T13:53:33","modified_gmt":"2025-07-18T13:53:33","slug":"hadoop-tutorial-series-issue-2-getting-started-with-customized-partitioning","status":"publish","type":"post","link":"https:\/\/philippeadjiman.com\/blog\/2009\/12\/20\/hadoop-tutorial-series-issue-2-getting-started-with-customized-partitioning\/","title":{"rendered":"Hadoop Tutorial Series, Issue #2: Getting Started With (Customized) Partitioning"},"content":{"rendered":"\n

In the Issue #1<\/a> of this series<\/a>, we set up the “learning playground” (based on the Cloudera Virtual Machine<\/a>) in order to enjoy hands-on learning experiences around Hadoop. In this issue, we’ll use our playground to investigate the partitioning features offered by Hadoop.<\/p>\n\n\n\n

What is it all about?<\/strong><\/p>\n\n\n\n

As you may know, a map\/reduce job will contains most of the time more than  1 reducer.  So basically, when a mapper emits a key value pair, it has to be sent to one of the reducers. Which one? The mechanism sending specific key-value pairs to specific reducers is called partitioning<\/strong> (the key-value pairs space is partitioned among the reducers). A Partitioner <\/a>is responsible to perform the partitioning.<\/p>\n\n\n\n

In Hadoop, the default partitioner is HashPartitioner<\/a>, which hashes a record’s key to determine which partition (and thus which reducer) the record belongs in.The number of partition is then equal to the number of reduce tasks for the job.<\/p>\n\n\n\n

Why is it important?<\/strong><\/p>\n\n\n\n

First, it has a direct impact on the overall performance of your job: a poorly designed partitioning function will not evenly distributes the charge over the reducers, potentially loosing all the interest of the map\/reduce distributed infrastructure.<\/p>\n\n\n\n

Second, it maybe sometimes necessary to control the key\/value pairs partitioning over the reducers. Let’s illustrate it on a simple example. Suppose that your job’s input is a (huge) set of tokens and their number of occurrences (for instance the output of the canonical word count hadoop example<\/a>) and that you want to sort them by number of occurrences. Let’s also suppose that your job will be handled by 2 reducers. If you run your job without using any customized partitioner, you’ll get something like this:<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n

As you can see, the tokens are correctly ordered by number of occurrences on each reducer (which is what hadoop guarantees by default) but this is not what you need! You’d rather expect something like:<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n

where tokens are totally ordered over the reducers, from 1 to 30 occurrences on the first reducer and from 31 to 14620 on the second. This would happen as a result of a correct partitioning function: all the tokens having a number of occurrences inferior to N (here 30) are sent  to reducer 1 and the others are sent to reducer 2, resulting in two partitions. Since the tokens are sorted on each partition, you get the expected total order on the number of occurrences.<\/p>\n\n\n\n

Below, we’ll use our playground to observe the issue happening  on real data and see how we solve it using customized partitioners.<\/p>\n\n\n\n

Also, as a second example of use of customized partitioning functions, let’s cite the original map\/reduce google paper<\/a>: “sometimes the output keys are URLs, and we want all entries for a single host to end up in the same output. To support situations like this, the user of the MapReduce library can provide a special partitioning function. For example, using “hash(Hostname(urlkey)) mod R” as the partitioning function causes all URLs from the same host to end up in the same output”.<\/p>\n\n\n\n

Feeling the partitions in our playground<\/strong><\/p>\n\n\n\n

If your playground is not yet set up, check the Issue #1 of this series<\/a>. As an input for our job, we’ll use a tsv file containing the list of tokens and their number of occurrences extracted from (once again<\/a>) the full moby dick text. Click here<\/a> to download this input. You’ll notice that the pairs (tokens, #occurrences) are alphanumerically sorted on tokens value.<\/p>\n\n\n\n

First, we’ll use a very simple pre-processing job to transform the input data into a more convenient format to use within hadoop: the Sequence File Output Format. Sequence files are a basic file based data structure persisting the key\/value pairs in a binary format and allowing you to interact more easily with basic hadoop data types (e.g IntWritable<\/a>, LongWritable<\/a>, etc…). Here is the simple pre-processing job:<\/p>\n\n\n\n

package com.philippeadjiman.hadooptraining;\n\nimport java.io.IOException;\n\nimport org.apache.hadoop.fs.Path;\nimport org.apache.hadoop.io.IntWritable;\nimport org.apache.hadoop.io.LongWritable;\nimport org.apache.hadoop.io.Text;\nimport org.apache.hadoop.mapred.FileInputFormat;\nimport org.apache.hadoop.mapred.FileOutputFormat;\nimport org.apache.hadoop.mapred.JobClient;\nimport org.apache.hadoop.mapred.JobConf;\nimport org.apache.hadoop.mapred.MapReduceBase;\nimport org.apache.hadoop.mapred.Mapper;\nimport org.apache.hadoop.mapred.OutputCollector;\nimport org.apache.hadoop.mapred.Reporter;\nimport org.apache.hadoop.mapred.SequenceFileOutputFormat;\n\npublic class SortDataPreprocessor {\n\n\tstatic class PreprocessorMapper extends MapReduceBase implements Mapper {\n\n\t\tprivate Text word = new Text();\n\n\t\tpublic void map(LongWritable key, Text value,\n\t\t\t\tOutputCollector output, Reporter reporter) throws IOException {\n\t\t\tString line = value.toString();\n\t\t\tString[] tokens = line.split(\"t\");\n\t\t\tif( tokens == null || tokens.length != 2 ){\n\t\t\t\tSystem.err.print(\"Problem with input line: \"+line+\"n\");\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tint nbOccurences = Integer.parseInt(tokens[1]);\n\t\t\tword.set(tokens[0]);\n\t\t\toutput.collect(new IntWritable(nbOccurences),word );\n\t\t}\n\t}\n\n\tpublic static void main(String[] args) throws IOException {\n\t\tJobConf conf = new JobConf(SortDataPreprocessor.class);\n\n\t\tFileInputFormat.setInputPaths(conf, new Path(args[0]));\n\t\tFileOutputFormat.setOutputPath(conf, new Path(args[1]));\n\n\t\tconf.setMapperClass(PreprocessorMapper.class);\n\t\tconf.setOutputKeyClass(IntWritable.class);\n\t\tconf.setOutputValueClass(Text.class);\n\t\tconf.setNumReduceTasks(0);\n\t\tconf.setOutputFormat(SequenceFileOutputFormat.class);\n\t\tJobClient.runJob(conf);\n\t}\n}<\/pre>\n\n\n\n
You’ll notice that:<\/p>\n\n\n\n