001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one
003     * or more contributor license agreements.  See the NOTICE file
004     * distributed with this work for additional information
005     * regarding copyright ownership.  The ASF licenses this file
006     * to you under the Apache License, Version 2.0 (the
007     * "License"); you may not use this file except in compliance
008     * with the License.  You may obtain a copy of the License at
009     *
010     *     http://www.apache.org/licenses/LICENSE-2.0
011     *
012     * Unless required by applicable law or agreed to in writing, software
013     * distributed under the License is distributed on an "AS IS" BASIS,
014     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015     * See the License for the specific language governing permissions and
016     * limitations under the License.
017     */
018    
019    package org.apache.hadoop.mapreduce.lib.join;
020    
021    import java.io.IOException;
022    import java.util.ArrayList;
023    import java.util.List;
024    import java.util.Map;
025    import java.util.regex.Matcher;
026    import java.util.regex.Pattern;
027    
028    import org.apache.hadoop.classification.InterfaceAudience;
029    import org.apache.hadoop.classification.InterfaceStability;
030    import org.apache.hadoop.conf.Configuration;
031    import org.apache.hadoop.fs.Path;
032    import org.apache.hadoop.io.WritableComparable;
033    import org.apache.hadoop.mapreduce.InputFormat;
034    import org.apache.hadoop.mapreduce.InputSplit;
035    import org.apache.hadoop.mapreduce.JobContext;
036    import org.apache.hadoop.mapreduce.RecordReader;
037    import org.apache.hadoop.mapreduce.TaskAttemptContext;
038    
039    /**
040     * An InputFormat capable of performing joins over a set of data sources sorted
041     * and partitioned the same way.
042     * @see #setFormat
043     *
044     * A user may define new join types by setting the property
045     * <tt>mapreduce.join.define.&lt;ident&gt;</tt> to a classname. 
046     * In the expression <tt>mapreduce.join.expr</tt>, the identifier will be
047     * assumed to be a ComposableRecordReader.
048     * <tt>mapreduce.join.keycomparator</tt> can be a classname used to compare 
049     * keys in the join.
050     * @see JoinRecordReader
051     * @see MultiFilterRecordReader
052     */
053    @SuppressWarnings("unchecked")
054    @InterfaceAudience.Public
055    @InterfaceStability.Stable
056    public class CompositeInputFormat<K extends WritableComparable>
057        extends InputFormat<K, TupleWritable> {
058    
059      public static final String JOIN_EXPR = "mapreduce.join.expr";
060      public static final String JOIN_COMPARATOR = "mapreduce.join.keycomparator";
061      
062      // expression parse tree to which IF requests are proxied
063      private Parser.Node root;
064    
065      public CompositeInputFormat() { }
066    
067    
068      /**
069       * Interpret a given string as a composite expression.
070       * {@code
071       *   func  ::= <ident>([<func>,]*<func>)
072       *   func  ::= tbl(<class>,"<path>")
073       *   class ::= @see java.lang.Class#forName(java.lang.String)
074       *   path  ::= @see org.apache.hadoop.fs.Path#Path(java.lang.String)
075       * }
076       * Reads expression from the <tt>mapreduce.join.expr</tt> property and
077       * user-supplied join types from <tt>mapreduce.join.define.&lt;ident&gt;</tt>
078       *  types. Paths supplied to <tt>tbl</tt> are given as input paths to the
079       * InputFormat class listed.
080       * @see #compose(java.lang.String, java.lang.Class, java.lang.String...)
081       */
082      public void setFormat(Configuration conf) throws IOException {
083        addDefaults();
084        addUserIdentifiers(conf);
085        root = Parser.parse(conf.get(JOIN_EXPR, null), conf);
086      }
087    
088      /**
089       * Adds the default set of identifiers to the parser.
090       */
091      protected void addDefaults() {
092        try {
093          Parser.CNode.addIdentifier("inner", InnerJoinRecordReader.class);
094          Parser.CNode.addIdentifier("outer", OuterJoinRecordReader.class);
095          Parser.CNode.addIdentifier("override", OverrideRecordReader.class);
096          Parser.WNode.addIdentifier("tbl", WrappedRecordReader.class);
097        } catch (NoSuchMethodException e) {
098          throw new RuntimeException("FATAL: Failed to init defaults", e);
099        }
100      }
101    
102      /**
103       * Inform the parser of user-defined types.
104       */
105      private void addUserIdentifiers(Configuration conf) throws IOException {
106        Pattern x = Pattern.compile("^mapreduce\\.join\\.define\\.(\\w+)$");
107        for (Map.Entry<String,String> kv : conf) {
108          Matcher m = x.matcher(kv.getKey());
109          if (m.matches()) {
110            try {
111              Parser.CNode.addIdentifier(m.group(1),
112                  conf.getClass(m.group(0), null, ComposableRecordReader.class));
113            } catch (NoSuchMethodException e) {
114              throw new IOException("Invalid define for " + m.group(1), e);
115            }
116          }
117        }
118      }
119    
120      /**
121       * Build a CompositeInputSplit from the child InputFormats by assigning the
122       * ith split from each child to the ith composite split.
123       */
124      @SuppressWarnings("unchecked")
125      public List<InputSplit> getSplits(JobContext job) 
126          throws IOException, InterruptedException {
127        setFormat(job.getConfiguration());
128        job.getConfiguration().setLong("mapreduce.input.fileinputformat.split.minsize", Long.MAX_VALUE);
129        return root.getSplits(job);
130      }
131    
132      /**
133       * Construct a CompositeRecordReader for the children of this InputFormat
134       * as defined in the init expression.
135       * The outermost join need only be composable, not necessarily a composite.
136       * Mandating TupleWritable isn't strictly correct.
137       */
138      @SuppressWarnings("unchecked") // child types unknown
139      public RecordReader<K,TupleWritable> createRecordReader(InputSplit split, 
140          TaskAttemptContext taskContext) 
141          throws IOException, InterruptedException {
142        setFormat(taskContext.getConfiguration());
143        return root.createRecordReader(split, taskContext);
144      }
145    
146      /**
147       * Convenience method for constructing composite formats.
148       * Given InputFormat class (inf), path (p) return:
149       * {@code tbl(<inf>, <p>) }
150       */
151      public static String compose(Class<? extends InputFormat> inf, 
152          String path) {
153        return compose(inf.getName().intern(), path, 
154                 new StringBuffer()).toString();
155      }
156    
157      /**
158       * Convenience method for constructing composite formats.
159       * Given operation (op), Object class (inf), set of paths (p) return:
160       * {@code <op>(tbl(<inf>,<p1>),tbl(<inf>,<p2>),...,tbl(<inf>,<pn>)) }
161       */
162      public static String compose(String op, 
163          Class<? extends InputFormat> inf, String... path) {
164        final String infname = inf.getName();
165        StringBuffer ret = new StringBuffer(op + '(');
166        for (String p : path) {
167          compose(infname, p, ret);
168          ret.append(',');
169        }
170        ret.setCharAt(ret.length() - 1, ')');
171        return ret.toString();
172      }
173    
174      /**
175       * Convenience method for constructing composite formats.
176       * Given operation (op), Object class (inf), set of paths (p) return:
177       * {@code <op>(tbl(<inf>,<p1>),tbl(<inf>,<p2>),...,tbl(<inf>,<pn>)) }
178       */
179      public static String compose(String op, 
180          Class<? extends InputFormat> inf, Path... path) {
181        ArrayList<String> tmp = new ArrayList<String>(path.length);
182        for (Path p : path) {
183          tmp.add(p.toString());
184        }
185        return compose(op, inf, tmp.toArray(new String[0]));
186      }
187    
188      private static StringBuffer compose(String inf, String path,
189          StringBuffer sb) {
190        sb.append("tbl(" + inf + ",\"");
191        sb.append(path);
192        sb.append("\")");
193        return sb;
194      }
195    }