Replicated Join Demo

Problem : How to do replicated join in Hadoop ?

Recently I learnt a new concept in Hadoop . By definition, Replicated Join means when one data set is too big and other data set is small enough that can be stored in memory , we can use Replicated Joins. Using this method, the join can be performed in the map side.

Here is the example of replicated join using Distributed Cache . I am using very simple xml files as my input data . Consider User data as small data that we will store in memory .


User Data

User Data

 













Comments Data










Please refer  link to create XmlInputFormat class . This is useful for reading XML data.
Now we can write Mapper Code. In the mapper's setup method, we are reading User xml file and storing in HashMap object.

public static class ReplicatedJoinMapper extends

   Mapper<LongWritable, Text, Text, Text> {

  private static final Log Logger = LogFactory

    .getLog(ReplicatedJoinMapper.class);

  private final String Empty_String = "null";


  private String joinType = null;

  private final HashMap<String, String> userInfo = new HashMap<String, String>();


  @Override

  protected void setup(Context context) throws java.io.IOException,

    InterruptedException {



   Path[] files = DistributedCache.getLocalCacheFiles(context

     .getConfiguration());



   if (files != null && files.length > 0) {

    try {

     if (files[0].getName().equals("User")) {

      // InputStream is = new

      // FileInputStream(files[0].toString());

      File is = new File(files[0].toString());


      DocumentBuilderFactory dbFactory = DocumentBuilderFactory

        .newInstance();


      DocumentBuilder dBuilder = dbFactory

        .newDocumentBuilder();

      Logger.info("-------------------------->2 = ----------->");

      Document doc = dBuilder.parse(is.toString().replaceAll(

        "[^\\x20-\\x7e\\x0A]", ""));



      doc.getDocumentElement().normalize();


      NodeList nList = doc.getElementsByTagName("employee");


      for (int temp = 0; temp < nList.getLength(); temp++) {


       Node nNode = nList.item(temp);


       if (nNode.getNodeType() == Node.ELEMENT_NODE) {


        Element eElement = (Element) nNode;


        String id = eElement.getElementsByTagName("id")

          .item(0).getTextContent();

      


        String name = eElement

          .getElementsByTagName("name").item(0)

          .getTextContent();

   

        String gender = eElement

          .getElementsByTagName("gender").item(0)

          .getTextContent();

   


        userInfo.put(id, id + "," + name + "," + gender);

        joinType = "inner";

       }

      }

     }


    } catch (Exception e) {

     // TODO: handle exception

     Logger.error(e.getMessage());

    }


   } else {

    Logger.info("-------------------------->file not found----------->");

   }

  }





// Here in map method we will read Comments XML file data


  @Override

  protected void map(LongWritable key, Text value, Context context)

    throws IOException, InterruptedException {

   try {


    InputStream is = new ByteArrayInputStream(value.toString()

      .getBytes());

    DocumentBuilderFactory dbFactory = DocumentBuilderFactory

      .newInstance();

    DocumentBuilder dBuilder = dbFactory.newDocumentBuilder();

    Document doc = dBuilder.parse(is);


    doc.getDocumentElement().normalize();


    NodeList nList = doc.getElementsByTagName("employee");


    for (int temp = 0; temp < nList.getLength(); temp++) {


     Node nNode = nList.item(temp);


     if (nNode.getNodeType() == Node.ELEMENT_NODE) {


      Element eElement = (Element) nNode;


      String id = eElement.getElementsByTagName("id").item(0)

        .getTextContent();

     


      String comment = eElement

        .getElementsByTagName("comment").item(0)

        .getTextContent();

    

      String userInfoLine = userInfo.get(id);

  

      if (userInfoLine != null) {


// Inner Join

       Logger.info("--------------------------> writing context----------->");

       context.write(new Text(id), new Text(userInfoLine

         + "," + comment));

      } else {


// Outer join

       context.write(new Text(id), new Text(Empty_String));

      }

     }

    }


   } catch (Exception e) {

    // TODO: handle exception

    Logger.error(e.getCause());

   }

  };

 }





Finally we can write Driver code. Here we are using DistributedCache.addCacheFile method to provide User xml to each map tasks. The same file is read in set up method above.





@Override

 public int run(String[] args) throws Exception {


  final Log Logger = LogFactory.getLog(RJDemo.class);


  Configuration conf = getConf();


  String[] arg = new GenericOptionsParser(conf, args).getRemainingArgs();


  


// These tags are in my xml file


  conf.set("START_TAG_KEY", "<employee>");

  conf.set("END_TAG_KEY", "</employee>");





  Job job = new Job(conf, "SO DEMO");


   try {

   DistributedCache.addCacheFile(new URI("/SODemoInput/User"),

   job.getConfiguration());

   } catch (Exception e) {

   e.printStackTrace();

   System.out.println(e);

   }

  job.setJarByClass(RJDemo.class);


  job.setMapperClass(ReplicatedJoinMapper.class);


  job.setNumReduceTasks(0);


  job.setInputFormatClass(XmlInputFormat.class);


  job.setOutputFormatClass(TextOutputFormat.class);


  // job.setOutputKeyClass(Text.class);


  job.setOutputValueClass(Text.class);


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


  Path out = new Path(args[1]);

  out.getFileSystem(conf).delete(out);


  FileInputFormat.addInputPath(job, in);


  FileOutputFormat.setOutputPath(job, out);


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


  return 0;

 }

Finally run this command to see output as:

hadoop jar  /home/Desktop/Projects/ReplicatedJoinDemo/RJDemo.jar RJDemo  /SODemoInput/Comments  /SODemoOutput/Out32



And the output looks like :

And that's it :)

 Have a good day ..
Happy Hadooping !!