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
019package org.apache.hadoop.mapreduce.lib.aggregate;
020
021import java.io.IOException;
022import java.util.Iterator;
023
024import org.apache.hadoop.classification.InterfaceAudience;
025import org.apache.hadoop.classification.InterfaceStability;
026import org.apache.hadoop.io.Text;
027import org.apache.hadoop.io.Writable;
028import org.apache.hadoop.io.WritableComparable;
029import org.apache.hadoop.mapreduce.Reducer;
030
031/**
032 * This class implements the generic combiner of Aggregate.
033 */
034@InterfaceAudience.Public
035@InterfaceStability.Stable
036public class ValueAggregatorCombiner<K1 extends WritableComparable<?>,
037                                     V1 extends Writable>
038  extends Reducer<Text, Text, Text, Text> {
039
040  /** Combines values for a given key.  
041   * @param key the key is expected to be a Text object, whose prefix indicates
042   * the type of aggregation to aggregate the values. 
043   * @param values the values to combine
044   * @param context to collect combined values
045   */
046  public void reduce(Text key, Iterable<Text> values, Context context) 
047      throws IOException, InterruptedException {
048    String keyStr = key.toString();
049    int pos = keyStr.indexOf(ValueAggregatorDescriptor.TYPE_SEPARATOR);
050    String type = keyStr.substring(0, pos);
051    long uniqCount = context.getConfiguration().
052      getLong(UniqValueCount.MAX_NUM_UNIQUE_VALUES, Long.MAX_VALUE);
053    ValueAggregator aggregator = ValueAggregatorBaseDescriptor
054      .generateValueAggregator(type, uniqCount);
055    for (Text val : values) {
056      aggregator.addNextValue(val);
057    }
058    Iterator<?> outputs = aggregator.getCombinerOutput().iterator();
059
060    while (outputs.hasNext()) {
061      Object v = outputs.next();
062      if (v instanceof Text) {
063        context.write(key, (Text)v);
064      } else {
065        context.write(key, new Text(v.toString()));
066      }
067    }
068  }
069}