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.fieldsel;
020    
021    import java.io.IOException;
022    import java.util.ArrayList;
023    import java.util.List;
024    
025    import org.apache.commons.logging.Log;
026    import org.apache.commons.logging.LogFactory;
027    import org.apache.hadoop.classification.InterfaceAudience;
028    import org.apache.hadoop.classification.InterfaceStability;
029    import org.apache.hadoop.conf.Configuration;
030    import org.apache.hadoop.io.Text;
031    import org.apache.hadoop.mapreduce.Reducer;
032    
033    /**
034     * This class implements a reducer class that can be used to perform field
035     * selections in a manner similar to unix cut. 
036     * 
037     * The input data is treated as fields separated by a user specified
038     * separator (the default value is "\t"). The user can specify a list of
039     * fields that form the reduce output keys, and a list of fields that form
040     * the reduce output values. The fields are the union of those from the key
041     * and those from the value.
042     * 
043     * The field separator is under attribute "mapreduce.fieldsel.data.field.separator"
044     * 
045     * The reduce output field list spec is under attribute 
046     * "mapreduce.fieldsel.reduce.output.key.value.fields.spec". 
047     * The value is expected to be like
048     * "keyFieldsSpec:valueFieldsSpec" key/valueFieldsSpec are comma (,) 
049     * separated field spec: fieldSpec,fieldSpec,fieldSpec ... Each field spec
050     * can be a simple number (e.g. 5) specifying a specific field, or a range
051     * (like 2-5) to specify a range of fields, or an open range (like 3-) 
052     * specifying all the fields starting from field 3. The open range field
053     * spec applies value fields only. They have no effect on the key fields.
054     * 
055     * Here is an example: "4,3,0,1:6,5,1-3,7-". It specifies to use fields
056     * 4,3,0 and 1 for keys, and use fields 6,5,1,2,3,7 and above for values.
057     */
058    @InterfaceAudience.Public
059    @InterfaceStability.Stable
060    public class FieldSelectionReducer<K, V>
061        extends Reducer<Text, Text, Text, Text> {
062    
063      private String fieldSeparator = "\t";
064    
065      private String reduceOutputKeyValueSpec;
066    
067      private List<Integer> reduceOutputKeyFieldList = new ArrayList<Integer>();
068    
069      private List<Integer> reduceOutputValueFieldList = new ArrayList<Integer>();
070    
071      private int allReduceValueFieldsFrom = -1;
072    
073      public static final Log LOG = LogFactory.getLog("FieldSelectionMapReduce");
074    
075      public void setup(Context context) 
076          throws IOException, InterruptedException {
077        Configuration conf = context.getConfiguration();
078        
079        this.fieldSeparator = 
080          conf.get(FieldSelectionHelper.DATA_FIELD_SEPERATOR, "\t");
081        
082        this.reduceOutputKeyValueSpec = 
083          conf.get(FieldSelectionHelper.REDUCE_OUTPUT_KEY_VALUE_SPEC, "0-:");
084        
085        allReduceValueFieldsFrom = FieldSelectionHelper.parseOutputKeyValueSpec(
086          reduceOutputKeyValueSpec, reduceOutputKeyFieldList,
087          reduceOutputValueFieldList);
088    
089        LOG.info(FieldSelectionHelper.specToString(fieldSeparator,
090          reduceOutputKeyValueSpec, allReduceValueFieldsFrom,
091          reduceOutputKeyFieldList, reduceOutputValueFieldList));
092      }
093    
094      public void reduce(Text key, Iterable<Text> values, Context context)
095          throws IOException, InterruptedException {
096        String keyStr = key.toString() + this.fieldSeparator;
097        
098        for (Text val : values) {
099          FieldSelectionHelper helper = new FieldSelectionHelper();
100          helper.extractOutputKeyValue(keyStr, val.toString(),
101            fieldSeparator, reduceOutputKeyFieldList,
102            reduceOutputValueFieldList, allReduceValueFieldsFrom, false, false);
103          context.write(helper.getKey(), helper.getValue());
104        }
105      }
106    }