{"id":21767,"date":"2014-02-19T22:00:22","date_gmt":"2014-02-19T20:00:22","guid":{"rendered":"http:\/\/www.javacodegeeks.com\/?p=21767"},"modified":"2014-02-19T15:21:27","modified_gmt":"2014-02-19T13:21:27","slug":"mapreduce-algorithms-understanding-data-joins-part-ii","status":"publish","type":"post","link":"https:\/\/www.javacodegeeks.com\/2014\/02\/mapreduce-algorithms-understanding-data-joins-part-ii.html","title":{"rendered":"MapReduce Algorithms – Understanding Data Joins Part II"},"content":{"rendered":"

It\u2019s been awhile since I last posted, and like last time I took a big break, I was taking some classes on Coursera. This time it was Functional Programming Principals in Scala<\/a> and Principles of Reactive Programming<\/a>. I found both of them to be great courses and would recommend taking either one if you have the time. In this post we resume our series on implementing the algorithms found in Data-Intensive Text Processing with MapReduce<\/a>, this time covering map-side joins. As we can guess from the name, map-side joins join data exclusively during the mapping phase and completely skip the reducing phase. In the last post on data joins we covered reduce side joins<\/a>. Reduce-side joins are easy to implement, but have the drawback that all data is sent across the network to the reducers. Map-side joins offer substantial gains in performance since we are avoiding the cost of sending data across the network. However, unlike reduce-side joins, map-side joins require very specific criteria be met. Today we will discuss the requirements for map-side joins and how we can implement them.<\/p>\n

Map-Side Join Conditions<\/h2>\n

To take advantage of map-side joins our data must meet one of following criteria:<\/p>\n

    \n
  1. The datasets to be joined are already sorted by the same key and have the same number of partitions<\/li>\n
  2. Of the two datasets to be joined, one is small enough to fit into memory<\/li>\n<\/ol>\n

    We are going to consider the first scenario where we have two (or more) datasets that need to be joined, but are too large to fit into memory. We will assume the worst case scenario, the files aren\u2019t sorted or partitioned the same.<\/p>\n

    Data Format<\/h2>\n

    Before we start, let\u2019s take a look at the data we are working with. We will have two datasets:<\/p>\n

      \n
    1. The first dataset consists of a GUID, First Name, Last Name, Address, City and State<\/li>\n
    2. The second dataset consists of a GUID and Employer information<\/li>\n<\/ol>\n

      Both datasets are comma delimited and the join-key (GUID) is in the first position. After the join we want the employer information from dataset two to be appended to the end of dataset one. Additionally, we want to keep the GUID in the first position of dataset one, but remove the GUID from dataset two.
      \nDataset 1:<\/p>\n

      aef9422c-d08c-4457-9760-f2d564d673bc,Linda,Narvaez,3253 Davis Street,Atlanta,GA\r\n  08db7c55-22ae-4199-8826-c67a5689f838,John,Gregory,258 Khale Street,Florence,SC\r\n  de68186a-1004-4211-a866-736f414eac61,Charles,Arnold,1764 Public Works Drive,Johnson City,TN\r\n  6df1882d-4c81-4155-9d8b-0c35b2d34284,John,Schofield,65 Summit Park Avenue,Detroit,MI<\/pre>\n
      Dataset 2:<\/p>\n
      de68186a-1004-4211-a866-736f414eac61,Jacobs\r\n  6df1882d-4c81-4155-9d8b-0c35b2d34284,Chief Auto Parts\r\n  aef9422c-d08c-4457-9760-f2d564d673bc,Earthworks Yard Maintenance\r\n  08db7c55-22ae-4199-8826-c67a5689f838,Ellman's Catalog Showrooms<\/pre>\n
      Joined results:<\/p>\n
      08db7c55-22ae-4199-8826-c67a5689f838,John,Gregory,258 Khale Street,Florence,SC,Ellman's Catalog Showrooms\r\n6df1882d-4c81-4155-9d8b-0c35b2d34284,John,Schofield,65 Summit Park Avenue,Detroit,MI,Chief Auto Parts\r\naef9422c-d08c-4457-9760-f2d564d673bc,Linda,Narvaez,3253 Davis Street,Atlanta,GA,Earthworks Yard Maintenance\r\nde68186a-1004-4211-a866-736f414eac61,Charles,Arnold,1764 Public Works Drive,Johnson City,TN,Jacobs<\/pre>\n
      Now we move on to how we go about joining our two datasets.<\/p>\n
      Map-Side Joins with Large Datasets<\/h2>\n
      To be able to perform map-side joins we need to have our data sorted by the same key and have the same number of partitions, implying that all keys for any record are in the same partition. While this seems to be a tough requirement, it is easily fixed. Hadoop sorts all keys and guarantees that keys with the same value are sent to the same reducer. So by simply running a MapReduce job that does nothing more than output the data by the key you want to join on and specifying the exact same number of reducers for all datasets, we will get our data in the correct form. Considering the gains in efficiency from being able to do a map-side join, it may be worth the cost of running additional MapReduce jobs. It bears repeating at this point it is crucial all datasets specify the exact<\/em> same number of reducers during the \u201cpreparation\u201d phase when the data will be sorted and partitioned. In this post we will take two data-sets and run an initial MapReduce job on both to do the sorting and partitioning and then run a final job to perform the map-side join. First let\u2019s cover the MapReduce job to sort and partition our data in the same way.<\/p>\n
      Step One: Sorting and Partitioning<\/h2>\n
      First we need to create a Mapper<\/code> that will simply choose the key for sorting by a given index:
       
      <\/div> <\/div><\/p>\n
      public class SortByKeyMapper extends Mapper<LongWritable, Text, Text, Text> {\r\n\r\n    private int keyIndex;\r\n    private Splitter splitter;\r\n    private Joiner joiner;\r\n    private Text joinKey = new Text();\r\n\r\n    @Override\r\n    protected void setup(Context context) throws IOException, InterruptedException {\r\n        String separator =  context.getConfiguration().get(\"separator\");\r\n        keyIndex = Integer.parseInt(context.getConfiguration().get(\"keyIndex\"));\r\n        splitter = Splitter.on(separator);\r\n        joiner = Joiner.on(separator);\r\n    }\r\n\r\n    @Override\r\n    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {\r\n        Iterable<String> values = splitter.split(value.toString());\r\n        joinKey.set(Iterables.get(values,keyIndex));\r\n        if(keyIndex != 0){\r\n            value.set(reorderValue(values,keyIndex));\r\n        }\r\n        context.write(joinKey,value);\r\n    }\r\n\r\n    private String reorderValue(Iterable<String> value, int index){\r\n        List<String> temp = Lists.newArrayList(value);\r\n        String originalFirst = temp.get(0);\r\n        String newFirst = temp.get(index);\r\n        temp.set(0,newFirst);\r\n        temp.set(index,originalFirst);\r\n        return joiner.join(temp);\r\n    }\r\n}<\/pre>\n
      The SortByKeyMapper<\/code> simply sets the value of the joinKey<\/code> by extracting the value from the given line of text found at the position given by the configuration parameter keyIndex<\/code>. Also, if the keyIndex<\/code> is not equal to zero, we swap the order of the values found in the first position and the keyIndex<\/code> position. Although this is a questionable feature, We\u2019ll discuss why we are doing this later. Next we need a Reducer<\/code>:<\/p>\n
      public class SortByKeyReducer extends Reducer<Text,Text,NullWritable,Text> {\r\n\r\n    private static final NullWritable nullKey = NullWritable.get();\r\n\r\n    @Override\r\n    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {\r\n        for (Text value : values) {\r\n             context.write(nullKey,value);\r\n        }\r\n    }\r\n}<\/pre>\n
      The SortByKeyReducer<\/code> writes out all values for the given key, but throws out the key and writes a NullWritable<\/code> instead. In the next section we will explain why we are not using the key.<\/p>\n
      Step Two: The Map-Side join<\/h2>\n
      When performing a map-side join the records are merged before<\/em> they reach the mapper. To achieve this, we use the CompositeInputFormat<\/a>. We will also need to set some configuration properties. Let\u2019s look at how we will configure our map-side join:<\/p>\n
      private static Configuration getMapJoinConfiguration(String separator, String... paths) {\r\n        Configuration config = new Configuration();\r\n        config.set(\"mapreduce.input.keyvaluelinerecordreader.key.value.separator\", separator);\r\n        String joinExpression = CompositeInputFormat.compose(\"inner\", KeyValueTextInputFormat.class, paths);\r\n        config.set(\"mapred.join.expr\", joinExpression);\r\n        config.set(\"separator\", separator);\r\n        return config;\r\n    }<\/pre>\n
      First, we are specifying the character that separates the key and values by setting the mapreduce.input.keyvaluelinerecordreader.key.value.separator<\/code> property. Next we use the CompositeInputFormat.compose<\/code> method to create a \u201cjoin expression\u201d specifying an inner join by using the word \u201cinner\u201d, then specifying the input format to use, the KeyValueTextInput<\/a>class and finally a String varargs representing the paths of the files to join (which are the output paths of the map-reduce jobs ran to sort and partition the data). The KeyValueTextInputFormat<\/code> class will use the separator character to set the first value as the key and the rest will be used for the value.<\/p>\n
      Mapper for the join<\/h2>\n
      Once the values from the source files have been joined, the Mapper.map<\/code> method is called, it will receive a Text<\/code> object for the key (the same key across joined records) and a TupleWritable<\/code> that is composed of the values joined from our input files for a given key. Remember we want our final output to have the join-key in the first position, followed by all of joined values in one delimited String<\/code>. To achieve this we have a custom mapper to put our data in the correct format:<\/p>\n
      public class CombineValuesMapper extends Mapper<Text, TupleWritable, NullWritable, Text> {\r\n\r\n    private static final NullWritable nullKey = NullWritable.get();\r\n    private Text outValue = new Text();\r\n    private StringBuilder valueBuilder = new StringBuilder();\r\n    private String separator;\r\n\r\n    @Override\r\n    protected void setup(Context context) throws IOException, InterruptedException {\r\n        separator = context.getConfiguration().get(\"separator\");\r\n    }\r\n\r\n    @Override\r\n    protected void map(Text key, TupleWritable value, Context context) throws IOException, InterruptedException {\r\n        valueBuilder.append(key).append(separator);\r\n        for (Writable writable : value) {\r\n            valueBuilder.append(writable.toString()).append(separator);\r\n        }\r\n        valueBuilder.setLength(valueBuilder.length() - 1);\r\n        outValue.set(valueBuilder.toString());\r\n        context.write(nullKey, outValue);\r\n        valueBuilder.setLength(0);\r\n    }\r\n}<\/pre>\n
      In the CombineValuesMapper<\/code> we are appending the key and all the joined values into one delimited String<\/code>. Here we can finally see the reason why we threw the join-key away in the previous MapReduce jobs. Since the key is the first position in the values for all the datasets to be joined, our mapper naturally eliminates the duplicate keys from the joined datasets. All we need to do is insert the given key into a StringBuilder<\/code>, then append the values contained in the TupleWritable<\/code>.<\/p>\n
      Putting It All Together<\/h2>\n
      Now we have all the code in place to run a map-side join on large datasets. Let\u2019s take a look at how we will run all the jobs together. As was stated before, we are assuming that our data is not sorted and partitioned the same, so we will need to run N (2 in this case) MapReduce jobs to get the data in the correct format. After the initial sorting\/partitioning jobs run, the final job performing the actual join will run.<\/p>\n
      public class MapSideJoinDriver {\r\n\r\n    public static void main(String[] args) throws Exception {\r\n        String separator = \",\";\r\n        String keyIndex = \"0\";\r\n        int numReducers = 10;\r\n        String jobOneInputPath = args[0];\r\n        String jobTwoInputPath = args[1];\r\n        String joinJobOutPath = args[2];\r\n\r\n        String jobOneSortedPath = jobOneInputPath + \"_sorted\";\r\n        String jobTwoSortedPath = jobTwoInputPath + \"_sorted\";\r\n\r\n        Job firstSort = Job.getInstance(getConfiguration(keyIndex, separator));\r\n        configureJob(firstSort, \"firstSort\", numReducers, jobOneInputPath, jobOneSortedPath, SortByKeyMapper.class, SortByKeyReducer.class);\r\n\r\n        Job secondSort = Job.getInstance(getConfiguration(keyIndex, separator));\r\n        configureJob(secondSort, \"secondSort\", numReducers, jobTwoInputPath, jobTwoSortedPath, SortByKeyMapper.class, SortByKeyReducer.class);\r\n\r\n        Job mapJoin = Job.getInstance(getMapJoinConfiguration(separator, jobOneSortedPath, jobTwoSortedPath));\r\n        configureJob(mapJoin, \"mapJoin\", 0, jobOneSortedPath + \",\" + jobTwoSortedPath, joinJobOutPath, CombineValuesMapper.class, Reducer.class);\r\n        mapJoin.setInputFormatClass(CompositeInputFormat.class);\r\n\r\n        List<Job> jobs = Lists.newArrayList(firstSort, secondSort, mapJoin);\r\n        int exitStatus = 0;\r\n        for (Job job : jobs) {\r\n            boolean jobSuccessful = job.waitForCompletion(true);\r\n            if (!jobSuccessful) {\r\n                System.out.println(\"Error with job \" + job.getJobName() + \"  \" + job.getStatus().getFailureInfo());\r\n                exitStatus = 1;\r\n                break;\r\n            }\r\n        }\r\n        System.exit(exitStatus);\r\n    }<\/pre>\n
      The MapSideJoinDriver<\/code> does the basic configuration for running MapReduce jobs. One interesting point is the sorting\/partitioning jobs specify 10 reducers each, while the final job explicitly sets the number of reducers to 0, since we are joining on the map-side and don\u2019t need a reduce phase. Since we don\u2019t have any complicated dependencies, we put the jobs in an ArrayList and run the jobs in linear order (lines 24-33).<\/p>\n
      Results<\/h2>\n
      Initially we had 2 files; name and address information in the first file and employment information in the second. Both files had a unique id in the first column.
      \nFile one:<\/p>\n
      ....\r\n08db7c55-22ae-4199-8826-c67a5689f838,John,Gregory,258 Khale Street,Florence,SC\r\n...<\/pre>\n
      File two:<\/p>\n
      ....\r\n08db7c55-22ae-4199-8826-c67a5689f838,Ellman's Catalog Showrooms\r\n....<\/pre>\n
      Results:<\/p>\n
      08db7c55-22ae-4199-8826-c67a5689f838,John,Gregory,258 Khale Street,Florence,SC,Ellman's Catalog Showrooms<\/pre>\n
      As we can see here, we\u2019ve successfully joined the records together and maintained the format of the files without duplicate keys in the results.<\/p>\n
      Conclusion<\/h2>\n
      In this post we\u2019ve demonstrated how to perform a map-side join when both data sets are large and can\u2019t fit into memory. If you get the feeling this takes a lot of work to pull off, you are correct. While in most cases we would want to use higher level tools like Pig or Hive, it\u2019s helpful to know the mechanics of performing map-side joins with large datasets. This especially true on those occasions when you need to write a solution from scratch. Thanks for your time.<\/p>\n
      Resources<\/h2>\n