title: Hadoop MultipleOutPutFormat and Join Query
body: <p>I am working on a hadoop task which was earlier populating few files in a directory say</p>
  
  <p>part-o
  part-1
  part-2</p>
  
  <p>I modified this task because of a requirement and used MultipleOutputs to capture few more outputs. so now directory structures looks like</p>
  
  <p>part-0
  part-1
  part-2
  output-1
  output-2
  output-3</p>
  
  <p>Issue : earlier few jobs were using this directory for a map side outer join but now the job has to take only part-* files for join and discard rest of the files.</p>
  
  <p>i tried giving the input as a ","  separated directories
  i.e <em>/part-1,</em>/part-2,*/part-3 </p>
  
  <p>and as per the following expression
  jobConf.set("mapred.join.expr", CompositeInputFormat.compose(outer,
                  KeyValueTextInputFormat.class, path[]))</p>
  
  <p>now my path[] contains 5 entires which ealier used to be 3, in a way the starting three indexes have <em>/part-1,</em>/part-2,*/part-3 paths and rest two as it were earlier.</p>
  
  <p>I am not sure if i am doing this right way, please suggest me what should i do so that this join works as it used to work earlier without output-* files.</p>
  
  <p>it is throwing following exception with the above approach.</p>
  
  <p>java.io.IOException: Inconsistent split cardinality from child 1 (12/6)
      org.apache.hadoop.mapred.join.Parser$CNode.getSplits(Parser.java:369)
      org.apache.hadoop.mapred.join.CompositeInputFormat.getSplits(CompositeInputFormat.java:117)
      org.apache.hadoop.mapred.JobClient.writeOldSplits(JobClient.java:810)
      org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:781)
      org.apache.hadoop.mapred.JobClient.submitJob(JobClient.java:730)</p>
  
  <p>JobConf Values are the following.</p>
  
  <pre><code> jobConf.setMapperClass(MyMapper.class);
      jobConf.setReducerClass(MyReducer.class);
  
    String[] foldersToJoin = StringUtils.split(getInputString(), Constants.COMMA);
      Path[] pathsToJoin =  new Path[foldersToJoin.length];
      int i = 0;
      for(String folder : foldersToJoin){
          pathsToJoin[i++] = new Path(folder);
      }
  
   FileOutputFormat.setOutputPath(jobConf, new Path("/MyOutPUT"));
  
      jobConf.setInputFormat(CompositeInputFormat.class);
      jobConf.set("mapred.join.expr", CompositeInputFormat.compose(Constants.OUTER_JOIN_OP,
              KeyValueTextInputFormat.class, pathsToJoin));
  
      jobConf.setOutputFormat(TextOutputFormat.class);
      jobConf.setOutputKeyClass(Text.class);
      jobConf.setOutputValueClass(Text.class);
  
      MultipleOutputs.addNamedOutput(jobConf, CHANGE_SET_A,
              TextOutputFormat.class, Text.class, Text.class);
      MultipleOutputs.addNamedOutput(jobConf, CHANGE_SET_B,
              TextOutputFormat.class, Text.class, Text.class);
      MultipleOutputs.addNamedOutput(jobConf,CHANGE_SET_C,
              TextOutputFormat.class, Text.class, Text.class);
      MultipleOutputs.addNamedOutput(jobConf, CHANGE_SET_D,
              TextOutputFormat.class, Text.class, Text.class);
  </code></pre>
  
  <p>i am appending all the files in the folder as a "," separated string following is the code to it</p>
  
  <pre><code>  mapred.join.expr = CompositeInputFormat.compose(Constants.OUTER_JOIN_OP,
          KeyValueTextInputFormat.class, pathsToJoin)
  
     where pathsToJoin =  new Path[]{new Path["/home/hadoop/folder1/part-1"],
    new Path["/home/hadoop/folder1/part-2"],new Path["/home/hadoop/folder1/part-3"],
     new Path["/home/hadoop/folder2"],new Path["/home/hadoop/folder3"] }
  </code></pre>
  
  <p>so basically i am trying to join only part-* files present in folder1 with folder2 and folder3</p>
  
  <p>Any doc, or any link to this kind of scenario would be of great help.</p>