HADOOP-1342. In aggregators, permit one to limit the number of unique values per key. Contributed by Runping.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@537926 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -11,6 +11,9 @@ Trunk (unreleased changes)
 
   3. HADOOP-1344.  Add RunningJob#getJobName(). (Michael Bieniosek via cutting)
 
+  4. HADOOP-1342.  In aggregators, permit one to limit the number of
+     unique values per key.  (Runping Qi via cutting)
+
 
 Branch 0.13 (unreleased changes)
 

src/java/org/apache/hadoop/mapred/lib/aggregate/UniqValueCount.java

@@ -29,16 +29,48 @@ import java.util.TreeMap;
  */
 public class UniqValueCount implements ValueAggregator {
 
-  TreeMap<Object, Object> uniqItems = null;
+  private TreeMap<Object, Object> uniqItems = null;
+
+  private long numItems = 0;
+  
+  private long maxNumItems = Long.MAX_VALUE;
 
   /**
    * the default constructor
    * 
    */
   public UniqValueCount() {
+    this(Long.MAX_VALUE);
+  }
+  
+  /**
+   * constructor
+   * @param maxNum the limit in the number of unique values to keep.
+   *  
+   */
+  public UniqValueCount(long maxNum) {
     uniqItems = new TreeMap<Object, Object>();
+    this.numItems = 0;
+    maxNumItems = Long.MAX_VALUE;
+    if (maxNum > 0 ) {
+      this.maxNumItems = maxNum;
+    }
   }
 
+  /**
+   * Set the limit on the number of unique values
+   * @param n the desired limit on the number of unique values
+   * @return the new limit on the number of unique values
+   */
+  public long setMaxItems(long n) {
+    if (n >= numItems) {
+      this.maxNumItems = n;
+    } else if (this.maxNumItems >= this.numItems) {
+      this.maxNumItems = this.numItems;
+    }
+    return this.maxNumItems;
+  }
+  
   /**
    * add a value to the aggregator
    * 
@@ -47,8 +79,10 @@ public class UniqValueCount implements ValueAggregator {
    * 
    */
   public void addNextValue(Object val) {
-    uniqItems.put(val, "1");
-
+    if (this.numItems <= this.maxNumItems) {
+      uniqItems.put(val, "1");
+      this.numItems = this.uniqItems.size();
+    }
   }
 
   /**

src/java/org/apache/hadoop/mapred/lib/aggregate/ValueAggregatorBaseDescriptor.java

@@ -46,7 +46,8 @@ public class ValueAggregatorBaseDescriptor implements ValueAggregatorDescriptor
   
   static public final String STRING_VALUE_MIN = "StringValueMin";
   
-
+  private static long maxNumItems = Long.MAX_VALUE;
+  
   public String inputFile = null;
 
   private static class MyEntry implements Entry {
@@ -106,7 +107,7 @@ public class ValueAggregatorBaseDescriptor implements ValueAggregatorDescriptor
     } else if (type.compareToIgnoreCase(DOUBLE_VALUE_SUM) == 0) {
       retv = new DoubleValueSum();
     } else if (type.compareToIgnoreCase(UNIQ_VALUE_COUNT) == 0) {
-      retv = new UniqValueCount();
+      retv = new UniqValueCount(maxNumItems);
     } else if (type.compareToIgnoreCase(VALUE_HISTOGRAM) == 0) {
       retv = new ValueHistogram();
     }
@@ -153,5 +154,6 @@ public class ValueAggregatorBaseDescriptor implements ValueAggregatorDescriptor
    */
   public void configure(JobConf job) {
     this.inputFile = job.get("map.input.file");
+    maxNumItems = job.getLong("aggregate.max.num.unique.values", Long.MAX_VALUE);
   }
 }

src/test/org/apache/hadoop/mapred/TestAggregates.java

@@ -69,7 +69,7 @@ public class TestAggregates extends TestCase {
     }
     expectedOutput.append("value_as_string_max\t9\n");
     expectedOutput.append("value_as_string_min\t1\n");
-    expectedOutput.append("uniq_count\t19\n");
+    expectedOutput.append("uniq_count\t15\n");
 
 
     fileOut.write(inputData.toString().getBytes("utf-8"));
@@ -96,6 +96,7 @@ public class TestAggregates extends TestCase {
     job.setInt("aggregator.descriptor.num", 1);
     job.set("aggregator.descriptor.0", 
           "UserDefined,org.apache.hadoop.mapred.lib.aggregate.AggregatorTests");
+    job.setLong("aggregate.max.num.unique.values", 14);
 
     JobClient.runJob(job);