title: MapReduce : Transform row-based data to a hierarchical form
body: <p>Follwing problem and code is from MapReduce design pattern.</p>
  
  <p>Problem: Given a list of posts and comments, create a structured XML hierarchy to nest comments with their related post.</p>
  
  <p>Throwing NullPointer Exception.I cannot catch my mistake. I really appreciate any help I can get.</p>
  
  <p>Here is my class and the nestElement method:</p>
  
  <pre><code>import java.io.IOException;
  import java.io.StringReader;
  import java.io.StringWriter;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  
  //import javax.swing.text.html.parser.Element;
  import javax.xml.parsers.DocumentBuilder;
  import javax.xml.parsers.DocumentBuilderFactory;
  import javax.xml.parsers.ParserConfigurationException;
  import javax.xml.transform.OutputKeys;
  import javax.xml.transform.Transformer;
  import javax.xml.transform.TransformerException;
  import javax.xml.transform.TransformerFactory;
  import javax.xml.transform.dom.DOMSource;
  import javax.xml.transform.stream.StreamResult;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.NullWritable;
  import org.apache.hadoop.io.Text;
  //import org.apache.hadoop.mapreduce.*;
  //import org.apache.hadoop.mapred.TextInputFormat;
  //import org.apache.hadoop.mapred.TextOutputFormat;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
  import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  //import org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer.Context;
  import org.w3c.dom.Attr;
  import org.w3c.dom.Document;
  import org.w3c.dom.NamedNodeMap;
  //import org.w3c.dom.UserDataHandler;
  import org.xml.sax.InputSource;
  import org.xml.sax.SAXException;
  public class PostCommentHierarchy {
          public static class PostMapper extends Mapper&lt;Object,Text,Text,Text&gt;{
                  private Text outkey = new Text();
                  private Text outvalue = new Text();
  
                  public void map(Object key,Text value,Context context)throws IOException ,InterruptedException{
                          Map&lt;String,String&gt; parsed = MRDPUtils.transformXmlToMap(value.toString());
                          String line = value.toString();
                          if(!(line.length()&lt;=0)){
                          outkey.set(parsed.get("Id"));
                          outvalue.set("P"+value.toString());
                          context.write(outkey, outvalue);
                          }
                  }
          }
  
          public static class CommentMapper extends Mapper&lt;Object,Text,Text,Text&gt;{
                  private Text outkey = new Text();
                  private Text outvalue = new Text();
  
                  public void map(Object key, Text value,Context context)throws IOException,InterruptedException{
                          Map&lt;String,String&gt; parsed = MRDPUtils.transformXmlToMap(value.toString());
                          String line = value.toString();
                          if(!(line.length()&lt;=0)){
                          outkey.set(parsed.get("PostId"));
  
                          outvalue.set("C"+value.toString());
                          context.write(outkey, outvalue);
                          }
                  }
          }
          public static class PostCommentHierarchyReducer extends Reducer &lt;Text,Text,Text,NullWritable&gt;{
                  private ArrayList&lt;String&gt; comments = new ArrayList&lt;String&gt;();
                  private DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
                  private String post = null;
  
                  public void reduce(Text key,Iterable&lt;Text&gt;values,Context context)throws IOException,InterruptedException{
                          post =null;
                          comments.clear();
  
                          for(Text t:values){
                              if(t.charAt(0)== 0){
                                  return;
                              }else if(t.charAt(0)=='P'){
                                          post = t.toString().substring(1,t.toString().length()).trim();
                                  }else{
                                          comments.add(t.toString().substring(1,t.toString().length()).trim());
                                  }
                          }
  
                          if(post != null){
                                  String postWithCommentChildren = null;
                                  try {
                                          try {
                                                  postWithCommentChildren = nestElements(post,comments);
                                          } catch (TransformerException e) {
                                                  // TODO Auto-generated catch block
                                                  e.printStackTrace();
                                          }
                                  } catch (ParserConfigurationException e) {
                                          // TODO Auto-generated catch block
                                          e.printStackTrace();
                                  } catch (SAXException e) {
                                          // TODO Auto-generated catch block
                                          e.printStackTrace();
                                  }
                                  context.write(new Text(postWithCommentChildren), NullWritable.get());
                          }
                  }
  
                  private String nestElements(String post,List&lt;String&gt; comments) throws ParserConfigurationException, SAXException, IOException, TransformerException{
                          DocumentBuilder bldr = dbf.newDocumentBuilder();
                          Document doc = bldr.newDocument();
  
                          org.w3c.dom.Element postEl = getXmlElementFromString(post);
                          org.w3c.dom.Element toAddPostEl = doc.createElement("post");
  //                      Element toAddPostEl1 = doc.createElement("post");
  
                          copyAttributesToElement(postEl.getAttributes(),toAddPostEl);
  
                          for(String commentXml:comments){
                                  org.w3c.dom.Element commentEl = getXmlElementFromString(commentXml);
                                  org.w3c.dom.Element toAddCommentEl = doc.createElement("comments");
  
                                  copyAttributesToElement(commentEl.getAttributes(),toAddCommentEl);
                                  toAddPostEl.appendChild(toAddPostEl);
                          }
                          doc.appendChild(toAddPostEl);
                      // Transform the document into a String of XML and return
                          return transformDocumentToString(doc);
                  }
                  private org.w3c.dom.Element getXmlElementFromString(String xml) throws ParserConfigurationException, SAXException, IOException{
                          DocumentBuilder bldr = dbf.newDocumentBuilder();
                          return bldr.parse(new InputSource(new StringReader(xml)))
                                          .getDocumentElement();
  
                  }
  
                  private void copyAttributesToElement(NamedNodeMap attributes,org.w3c.dom.Element toAddPostEl){
                          for(int i = 0; i&lt;attributes.getLength();++i){
                                  Attr toCopy = (Attr) attributes.item(i);
                                  toAddPostEl.setAttribute(toCopy.getName(),toCopy.getValue());
  
                          }
                  }
  
                  private String transformDocumentToString(Document doc) throws TransformerException{
  
                          TransformerFactory tf = TransformerFactory.newInstance();
                          Transformer transformer = tf.newTransformer();
                          transformer.setOutputProperty(OutputKeys.OMIT_XML_DECLARATION, "yes");
  
                          StringWriter writer = new StringWriter();
                          //Acts as a holder for a transformation Source tree in the form of a Document Object Model (DOM) tree.
                          transformer.transform(new DOMSource(doc), new StreamResult(writer));
                          return writer.getBuffer().toString().replaceAll("\n|\r", "");
                  }
  
          }
          public static void main(String[] args)throws IOException, ClassNotFoundException, InterruptedException{
  
                  //check how multiple inputs job works?
  
                  Configuration conf = new Configuration();
                  @SuppressWarnings("deprecation")
                  Job job = new Job(conf,"PostCommentHierarchy");
                  job.setJarByClass(PostCommentHierarchy.class);
                  job.setOutputKeyClass(Text.class);
                  job.setOutputValueClass(Text.class);
  
                  MultipleInputs.addInputPath(job, new Path(args[0]),TextInputFormat.class,PostMapper.class);
                  MultipleInputs.addInputPath(job, new Path(args[1]),TextInputFormat.class ,CommentMapper.class);
  
                  job.setReducerClass(PostCommentHierarchyReducer.class);
                  job.setOutputFormatClass(TextOutputFormat.class);
                  TextOutputFormat.setOutputPath(job, new Path(args[2]));
  
  
                  System.exit(job.waitForCompletion(true)? 0 :1);
          }
  
  }
  </code></pre>