You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2011/11/17 20:39:05 UTC
svn commit: r1203345 - in /incubator/accumulo/branches/1.4: docs/
docs/examples/ docs/src/user_manual/chapters/
src/core/src/main/java/org/apache/accumulo/core/iterators/
src/core/src/main/java/org/apache/accumulo/core/iterators/conf/
src/core/src/main...
Author: kturner
Date: Thu Nov 17 19:39:04 2011
New Revision: 1203345
URL: http://svn.apache.org/viewvc?rev=1203345&view=rev
Log:
ACCUMULO-148 ACCUMULO-154
Added:
incubator/accumulo/branches/1.4/docs/combiners.html
- copied, changed from r1203267, incubator/accumulo/branches/1.4/docs/aggregation.html
incubator/accumulo/branches/1.4/docs/examples/README.combiner
- copied, changed from r1202876, incubator/accumulo/branches/1.4/docs/examples/README.aggregation
incubator/accumulo/branches/1.4/src/examples/src/main/java/org/apache/accumulo/examples/combiner/
incubator/accumulo/branches/1.4/src/examples/src/main/java/org/apache/accumulo/examples/combiner/StatsCombiner.java
Removed:
incubator/accumulo/branches/1.4/docs/aggregation.html
incubator/accumulo/branches/1.4/docs/examples/README.aggregation
incubator/accumulo/branches/1.4/src/examples/src/main/java/org/apache/accumulo/examples/aggregation/
Modified:
incubator/accumulo/branches/1.4/docs/index.html
incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/analytics.tex
incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/table_configuration.tex
incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java
incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/ColumnSet.java
incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/PerColumnIteratorConfig.java
incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/util/shell/commands/SetIterCommand.java
incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java
incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTest.java
incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/test/functional/ServerSideErrorTest.java
Copied: incubator/accumulo/branches/1.4/docs/combiners.html (from r1203267, incubator/accumulo/branches/1.4/docs/aggregation.html)
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/docs/combiners.html?p2=incubator/accumulo/branches/1.4/docs/combiners.html&p1=incubator/accumulo/branches/1.4/docs/aggregation.html&r1=1203267&r2=1203345&rev=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/docs/aggregation.html (original)
+++ incubator/accumulo/branches/1.4/docs/combiners.html Thu Nov 17 19:39:04 2011
@@ -16,68 +16,69 @@
-->
<html>
<head>
-<title>Accumulo Documentation : Aggregation</title>
+<title>Accumulo Documentation : Combiners</title>
<link rel='stylesheet' type='text/css' href='documentation.css' media='screen'/>
</head>
<body>
-<h1>Aggregation</h1>
+<h1>Combiners</h1>
-<p>Accumulo does aggregation differently than traditional RDBMSs. Instead of aggregating at query time, it aggregates at ingest time. Doing aggregation at ingest time is ideal for large amounts of data and has three distinct advantages. First, it reduces the actual amount of data stored. Second, it makes queries run faster. Third, it removes the need to do a lookup at insert time in many cases, which can greatly speed up ingest.
+<p>Accumulo supports on the fly lazy aggregation of data using Combiners. Aggregation is done at compaction and scan time. No lookup is done at insert time, which` greatly speeds up ingest.
-<p>Aggregation in accumulo is easy to use. You simply specify which columns or column family you want to aggregate at table creation time. Allowing an aggregation function to apply to a whole column family is an interesting twist that gives the user great flexibility. The example below demonstrates this flexibility.
+<p>Combiners are easy to use. You use the setiters command to configure a combiner for a table. Allowing a Combiner to apply to a whole column family is an interesting twist that gives the user great flexibility. The example below demonstrates this flexibility.
<p><pre>
+
Shell - Accumulo Interactive Shell
- version: 1.4.0-incubating-SNAPSHOT
- instance id: 863fc0d1-3623-4b6c-8c23-7d4fdb1c8a49
-
- type 'help' for a list of available commands
--
-user@instance:9999> createtable perDayCounts -a day=org.apache.accumulo.core.iterators.aggregation.StringSummation
-user@instance:9999 perDayCounts> insert foo day 20080101 1
-insert successful
-user@instance:9999 perDayCounts> insert foo day 20080101 1
-insert successful
-user@instance:9999 perDayCounts> insert foo day 20080103 1
-insert successful
-user@instance:9999 perDayCounts> insert bar day 20080101 1
-insert successful
-user@instance:9999 perDayCounts> insert bar day 20080101 1
-insert successful
-user@instance:9999 perDayCounts> scan
-bar day:20080101 [] 2
-foo day:20080101 [] 2
-foo day:20080103 [] 1
-user@instance:9999 perDayCounts>
+-
+user@instance> createtable perDayCounts
+user@instance perDayCounts> setiter -t perDayCounts -p 10 -scan -minc -majc -n daycount -class org.apache.accumulo.core.iterators.user.SummingCombiner
+TypedValueCombiner can interpret Values as a variety of number encodings (VLong, Long, or String) before combining
+----------> set SummingCombiner parameter columns, <col fam>[:<col qual>]{,<col fam>[:<col qual>]} escape non aplhanum chars using %<hex>.: day
+----------> set SummingCombiner parameter type, <VARNUM|LONG|STRING>: STRING
+user@instance perDayCounts> insert foo day 20080101 1
+user@instance perDayCounts> insert foo day 20080101 1
+user@instance perDayCounts> insert foo day 20080103 1
+user@instance perDayCounts> insert bar day 20080101 1
+user@instance perDayCounts> insert bar day 20080101 1
+user@instance perDayCounts> scan
+bar day:20080101 [] 2
+foo day:20080101 [] 2
+foo day:20080103 [] 1
</pre>
-<p>Implementing a new aggregation function is a snap. Simply write some Java code that implements <a href='apidocs/org/apache/accumulo/core/iterators/aggregation/Aggregator.html'>org.apache.accumulo.core.iterators.aggregation.Aggregator</a>. A good example to look at is <a href='apidocs/org/apache/accumulo/core/iterators/aggregation/StringSummation.html'>StringSummation</a> which sums numbers encoded as ascii strings. However, one could easily write a much more efficient summation aggregator that operates on numbers encoded in twos complement.
-
-<p>To deploy a new aggregator, jar it up and put the jar in accumulo/lib. To see an example look at <a href='examples/README.aggregation'>README.aggregation</a>
+<p>Implementing a new Combiner is a snap. Simply write some Java code that extends <a href='apidocs/org/apache/accumulo/core/iterators/Combiner.html'>org.apache.accumulo.core.iterators.Combiner</a>. A good place to look for examples is the <a href='apidocs/org/apache/accumulo/core/iterators/user/package-summary.html'>org.apache.accumulo.core.iterators.user</a> package. Also look at the example StatsCombiner.
-<p>If you would like to see what aggregators a table has you can use the config command like in the following example.
-
-<p><pre>
-user@instance:9999 perDayCounts> config -t perDayCounts agg
----------+------------------------------------------+-----------------------------------------
-SCOPE | NAME | VALUE
----------+------------------------------------------+-----------------------------------------
-table | table.iterator.majc.agg................. | 10,org.apache.accumulo.core.iterators.AggregatingIterator
-table | table.iterator.majc.agg.opt.day......... | org.apache.accumulo.core.iterators.aggregation.StringSummation
-table | table.iterator.minc.agg................. | 10,org.apache.accumulo.core.iterators.AggregatingIterator
-table | table.iterator.minc.agg.opt.day......... | org.apache.accumulo.core.iterators.aggregation.StringSummation
-table | table.iterator.scan.agg................. | 10,org.apache.accumulo.core.iterators.AggregatingIterator
-table | table.iterator.scan.agg.opt.day......... | org.apache.accumulo.core.iterators.aggregation.StringSummation
----------+------------------------------------------+-----------------------------------------
-user@instance:9999 perDayCounts>
-</pre>
+<p>To deploy a new aggregator, jar it up and put the jar in accumulo/lib/ext. To see an example look at <a href='examples/README.combiner'>README.combiner</a>
-<p>You can add aggregators to an existing table using the following command in the accumulo shell.
+<p>If you would like to see what iterators a table has you can use the config command like in the following example.
<p><pre>
- config -t <tablename> <columnFamily[:columnQual]>=<aggregation class>
+user@instance perDayCounts> config -t perDayCounts -f iterator
+---------+---------------------------------------------+-----------------------------------------------------------
+SCOPE | NAME | VALUE
+---------+---------------------------------------------+-----------------------------------------------------------
+table | table.iterator.majc.daycount .............. | 10,org.apache.accumulo.core.iterators.user.SummingCombiner
+table | table.iterator.majc.daycount.opt.columns .. | day
+table | table.iterator.majc.daycount.opt.type ..... | STRING
+table | table.iterator.majc.vers .................. | 20,org.apache.accumulo.core.iterators.VersioningIterator
+table | table.iterator.majc.vers.opt.maxVersions .. | 1
+table | table.iterator.minc.daycount .............. | 10,org.apache.accumulo.core.iterators.user.SummingCombiner
+table | table.iterator.minc.daycount.opt.columns .. | day
+table | table.iterator.minc.daycount.opt.type ..... | STRING
+table | table.iterator.minc.vers .................. | 20,org.apache.accumulo.core.iterators.VersioningIterator
+table | table.iterator.minc.vers.opt.maxVersions .. | 1
+table | table.iterator.scan.daycount .............. | 10,org.apache.accumulo.core.iterators.user.SummingCombiner
+table | table.iterator.scan.daycount.opt.columns .. | day
+table | table.iterator.scan.daycount.opt.type ..... | STRING
+table | table.iterator.scan.vers .................. | 20,org.apache.accumulo.core.iterators.VersioningIterator
+table | table.iterator.scan.vers.opt.maxVersions .. | 1
+---------+---------------------------------------------+-----------------------------------------------------------
</pre>
</body>
Copied: incubator/accumulo/branches/1.4/docs/examples/README.combiner (from r1202876, incubator/accumulo/branches/1.4/docs/examples/README.aggregation)
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/docs/examples/README.combiner?p2=incubator/accumulo/branches/1.4/docs/examples/README.combiner&p1=incubator/accumulo/branches/1.4/docs/examples/README.aggregation&r1=1202876&r2=1203345&rev=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/docs/examples/README.aggregation (original)
+++ incubator/accumulo/branches/1.4/docs/examples/README.combiner Thu Nov 17 19:39:04 2011
@@ -1,4 +1,4 @@
-Title: Aggregation Example
+Title: Combiner Example
Notice: Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
@@ -16,7 +16,7 @@ Notice: Licensed to the Apache Softwa
specific language governing permissions and limitations
under the License.
-This is a simple aggregation example. To build this example run maven and then
+This is a simple combiner example. To build this example run maven and then
copy the produced jar into the accumulo lib dir. This is already done in the
tar distribution.
@@ -31,20 +31,34 @@ tar distribution.
-
- type 'help' for a list of available commands
-
- username@instance> createtable aggtest1 -a app=org.apache.accumulo.examples.aggregation.SortedSetAggregator
- username@instance aggtest1> insert foo app 1 a
- username@instance aggtest1> insert foo app 1 b
- username@instance aggtest1> scan
- foo app:1 [] a,b
- username@instance aggtest1> insert foo app 1 z,1,foo,w
- username@instance aggtest1> scan
- foo app:1 [] 1,a,b,foo,w,z
- username@instance aggtest1> insert foo app 2 cat,dog,muskrat
- username@instance aggtest1> insert foo app 2 mouse,bird
- username@instance aggtest1> scan
- foo app:1 [] 1,a,b,foo,w,z
- foo app:2 [] bird,cat,dog,mouse,muskrat
- username@instance aggtest1>
+ username@instance> createtable runners
+ username@instance runners> setiter -t runners -p 10 -scan -minc -majc -n decStats -class org.apache.accumulo.examples.combiner.StatsCombiner
+ Combiner that keeps track of min, max, sum, and count
+ ----------> set StatsCombiner parameter columns, <col fam>[:<col qual>]{,<col fam>[:<col qual>]} escape non aplhanum chars using %<hex>.: stat
+ ----------> set StatsCombiner parameter radix, radix/base of the numbers: 10
+ username@instance runners> setiter -t runners -p 10 -scan -minc -majc -n hexStats -class org.apache.accumulo.examples.combiner.StatsCombiner
+ Combiner that keeps track of min, max, sum, and count
+ ----------> set StatsCombiner parameter columns, <col fam>[:<col qual>]{,<col fam>[:<col qual>]} escape non aplhanum chars using %<hex>.: hstat
+ ----------> set StatsCombiner parameter radix, radix/base of the numbers: 16
+ username@instance runners> insert 123456 name first Joe
+ username@instance runners> insert 123456 stat marathon 240
+ username@instance runners> scan
+ 123456 name:first [] Joe
+ 123456 stat:marathon [] 240,240,240,1
+ username@instance runners> insert 123456 stat marathon 230
+ username@instance runners> insert 123456 stat marathon 220
+ username@instance runners> scan
+ 123456 name:first [] Joe
+ 123456 stat:marathon [] 220,240,690,3
+ username@instance runners> insert 123456 hstat virtualMarathon 6a
+ username@instance runners> insert 123456 hstat virtualMarathon 6b
+ username@instance runners> scan
+ 123456 hstat:virtualMarathon [] 6a,6b,d5,2
+ 123456 name:first [] Joe
+ 123456 stat:marathon [] 220,240,690,3
-In this example a table is created and the example set aggregator is
-applied to the column family app.
+In this example a table is created and the example stats combiner is applied to
+the column family stat and hstat. The stats combiner computes min,max,sum, and
+count. It can be configured to use a different base or radix. In the example
+above the column family stat is configured for base 10 and the column family
+hstat is configured for base 16.
Modified: incubator/accumulo/branches/1.4/docs/index.html
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/docs/index.html?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/docs/index.html (original)
+++ incubator/accumulo/branches/1.4/docs/index.html Thu Nov 17 19:39:04 2011
@@ -25,7 +25,7 @@
<ul>
<li><a href=accumulo_user_manual.pdf>User Manual</a></li>
<li><a href=administration.html>Administration</a></li>
-<li><a href=aggregation.html>Aggregation</a></li>
+<li><a href=combiners.html>Combiners</a></li>
<li><a href=constraints.html>Constraints</a></li>
<li><a href=bulkIngest.html>Bulk Ingest</a></li>
<li><a href=config.html>Configuration</a></li>
Modified: incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/analytics.tex
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/analytics.tex?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/analytics.tex (original)
+++ incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/analytics.tex Thu Nov 17 19:39:04 2011
@@ -154,33 +154,33 @@ AccumuloOutputFormat.setMaxMutationBuffe
An example of using MapReduce with Accumulo can be found at\\
accumulo/docs/examples/README.mapred
-\section{Aggregating Iterators}
+\section{Combiners}
Many applications can benefit from the ability to aggregate values across common
-keys. This can be done via aggregating iterators and is similar to the Reduce step in
+keys. This can be done via Combiner iterators and is similar to the Reduce step in
MapReduce. This provides the ability to define online, incrementally updated
analytics without the overhead or latency associated with batch-oriented
MapReduce jobs.
All that is needed to aggregate values of a table is to identify the fields over which
values will be grouped, insert mutations with those fields as the key, and configure
-the table with an aggregating iterator that supports the summarization operation
+the table with a combining iterator that supports the summarization operation
desired.
-The only restriction on an aggregating iterator is that the aggregator developer
+The only restriction on an combining iterator is that the combiner developer
should not assume that all values for a given key have been seen, since new
mutations can be inserted at anytime. This precludes using the total number of
values in the aggregation such as when calculating an average, for example.
\subsection{Feature Vectors}
-An interesting use of aggregating iterators within a Accumulo table is to store
+An interesting use of combining iterators within a Accumulo table is to store
feature vectors for use in machine learning algorithms. For example, many
algorithms such as k-means clustering, support vector machines, anomaly detection,
etc. use the concept of a feature vector and the calculation of distance metrics to
learn a particular model. The columns in a Accumulo table can be used to efficiently
store sparse features and their weights to be incrementally updated via the use of an
-aggregating iterator.
+combining iterator.
\section{Statistical Modeling}
Modified: incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/table_configuration.tex
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/table_configuration.tex?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/table_configuration.tex (original)
+++ incubator/accumulo/branches/1.4/docs/src/user_manual/chapters/table_configuration.tex Thu Nov 17 19:39:04 2011
@@ -293,15 +293,15 @@ table | table.iterator.scan.vers.opt.
\end{verbatim}
\normalsize
-\subsection{Aggregating Iterators}
+\subsection{Combiners}
-Accumulo allows aggregating iterators to be configured on tables and column
-families. When an aggregating iterator is set, the iterator is applied across the values
+Accumulo allows Combiners to be configured on tables and column
+families. When a Combiner is set it is applied across the values
associated with any keys that share rowID, column family, and column qualifier.
This is similar to the reduce step in MapReduce, which applied some function to all
the values associated with a particular key.
-For example, if an aggregating iterator were configured on a table and the following
+For example, if a summing combiner were configured on a table and the following
mutations were inserted:
\small
@@ -320,40 +320,40 @@ rowID1 colfA colqA - 2
\end{verbatim}
\normalsize
-Aggregating iterators can be enabled for a table as follows:
+Combiners can be enabled for a table using the setiter command in the shell. Below is an example.
\small
\begin{verbatim}
-user@myinstance> createtable perDayCounts -a
-day=org.apache.accumulo.core.iterators.aggregation.StringSummation
+root@a14 perDayCounts> setiter -t perDayCounts -p 10 -scan -minc -majc -n daycount
+ -class org.apache.accumulo.core.iterators.user.SummingCombiner
+TypedValueCombiner can interpret Values as a variety of number encodings
+ (VLong, Long, or String) before combining
+----------> set SummingCombiner parameter columns,
+ <col fam>[:<col qual>]{,<col fam>[:<col qual>]} : day
+----------> set SummingCombiner parameter type, <VARNUM|LONG|STRING>: STRING
-user@myinstance perDayCounts> insert row1 day 20080101 1
-user@myinstance perDayCounts> insert row1 day 20080101 1
-user@myinstance perDayCounts> insert row1 day 20080103 1
-user@myinstance perDayCounts> insert row2 day 20080101 1
-user@myinstance perDayCounts> insert row3 day 20080101 1
+root@a14 perDayCounts> insert foo day 20080101 1
+root@a14 perDayCounts> insert foo day 20080101 1
+root@a14 perDayCounts> insert foo day 20080103 1
+root@a14 perDayCounts> insert bar day 20080101 1
+root@a14 perDayCounts> insert bar day 20080101 1
-user@myinstance perDayCounts> scan
-row1 day:20080101 [] 2
-row1 day:20080103 [] 1
-row2 day:20080101 [] 2
+root@a14 perDayCounts> scan
+bar day:20080101 [] 2
+foo day:20080101 [] 2
+foo day:20080103 [] 1
\end{verbatim}
\normalsize
-Accumulo includes the following aggregators:
-\begin{itemize}
-\item{\textbf{LongSummation}: expects values of type long and adds them.}
-\item{\textbf{StringSummation}: expects numbers represented as strings and adds them.}
-\item{\textbf{StringMax}: expects numbers as strings and retains the maximum number inserted.}
-\item{\textbf{StringMin}: expects numbers as strings and retains the minimum number inserted.}
-\end{itemize}
+Accumulo includes some useful Combiners out of the box. To find these look in
+the\\ \textbf{org.apache.accumulo.core.iterators.user} package.
-Additional Aggregators can be added by creating a Java class that implements\\
-\textbf{org.apache.accumulo.core.iterators.aggregation.Aggregator} and adding a jar containing that
-class to Accumulo's lib directory.
+Additional Combiners can be added by creating a Java class that extends\\
+\textbf{org.apache.accumulo.core.iterators.Combiner} and adding a jar containing that
+class to Accumulo's lib/ext directory.
-An example of an aggregator can be found under\\
-accumulo/src/examples/main/java/org/apache/accumulo/examples/aggregation/SortedSetAggregator.java
+An example of a Combiner can be found under\\
+accumulo/src/examples/main/java/org/apache/accumulo/examples/combiner/StatsCombiner.java
\section{Block Cache}
Modified: incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java Thu Nov 17 19:39:04 2011
@@ -17,13 +17,12 @@
package org.apache.accumulo.core.iterators;
import java.io.IOException;
-import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.NoSuchElementException;
import org.apache.accumulo.core.client.IteratorSetting;
@@ -33,24 +32,23 @@ import org.apache.accumulo.core.data.Par
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.conf.ColumnSet;
-import org.apache.accumulo.core.iterators.conf.PerColumnIteratorConfig;
+import org.apache.accumulo.core.util.Pair;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
-/*
- * A SortedKeyValueIterator that combines the Values for different versions of a Key into a single Value.
- * Combiner will replace one or more versions of a Key and their Values with the most recent Key and a Value which is the result of the reduce method.
+/**
+ * A SortedKeyValueIterator that combines the Values for different versions of a Key into a single Value. Combiner will replace one or more versions of a Key
+ * and their Values with the most recent Key and a Value which is the result of the reduce method.
*
- * Subclasses must implement a reduce method:
- * public Value reduce(Key key, Iterator<Value> iter);
+ * Subclasses must implement a reduce method: public Value reduce(Key key, Iterator<Value> iter);
*
* This reduce method will be passed the most recent Key and an iterator over the Values for all non-deleted versions of that Key.
*/
public abstract class Combiner extends WrappingIterator implements OptionDescriber {
static final Logger log = Logger.getLogger(Combiner.class);
- public static final String COLUMN_PREFIX = "column:";
+ public static final String COLUMNS_OPTION = "columns";
- /*
+ /**
* A Java Iterator that iterates over the Values for a given Key from a source SortedKeyValueIterator.
*/
public static class ValueIterator implements Iterator<Value> {
@@ -58,10 +56,11 @@ public abstract class Combiner extends W
SortedKeyValueIterator<Key,Value> source;
boolean hasNext;
- /*
+ /**
* Constructs an iterator over Values whose Keys are versions of the current topKey of the source SortedKeyValueIterator.
*
- * @param source The SortedKeyValueIterator<Key,Value> from which to read data.
+ * @param source
+ * The SortedKeyValueIterator<Key,Value> from which to read data.
*/
public ValueIterator(SortedKeyValueIterator<Key,Value> source) {
this.source = source;
@@ -73,7 +72,7 @@ public abstract class Combiner extends W
return source.hasTop() && !source.getTopKey().isDeleted() && topKey.equals(source.getTopKey(), PartialKey.ROW_COLFAM_COLQUAL_COLVIS);
}
- /*
+ /**
* @return <tt>true</tt> if there is another Value
*
* @see java.util.Iterator#hasNext()
@@ -83,7 +82,7 @@ public abstract class Combiner extends W
return hasNext;
}
- /*
+ /**
* @return the next Value
*
* @see java.util.Iterator#next()
@@ -102,7 +101,7 @@ public abstract class Combiner extends W
return topValue;
}
- /*
+ /**
* unsupported
*
* @see java.util.Iterator#remove()
@@ -151,7 +150,7 @@ public abstract class Combiner extends W
private Key workKey = new Key();
- /*
+ /**
* Sets the topKey and topValue based on the top key of the source. If the column of the source top key is in the set of combiners, topKey will be the top key
* of the source and topValue will be the result of the reduce method. Otherwise, topKey and topValue will be unchanged. (They are always set to null before
* this method is called.)
@@ -195,12 +194,14 @@ public abstract class Combiner extends W
}
}
- /*
+ /**
* Reduces a list of Values into a single Value.
*
- * @param key The most recent version of the Key being reduced.
+ * @param key
+ * The most recent version of the Key being reduced.
*
- * @param iter An iterator over the Values for different versions of the key.
+ * @param iter
+ * An iterator over the Values for different versions of the key.
*
* @return The combined Value.
*/
@@ -212,36 +213,87 @@ public abstract class Combiner extends W
public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
super.init(source, options, env);
- List<String> colOpts = new ArrayList<String>();
- for (Entry<String,String> opt : options.entrySet()) {
- if ((opt.getKey().startsWith(COLUMN_PREFIX))) {
- colOpts.add(opt.getKey().substring(COLUMN_PREFIX.length()));
- }
- }
- combiners = new ColumnSet(colOpts);
+ if (!options.containsKey(COLUMNS_OPTION))
+ throw new IllegalArgumentException("Must specify " + COLUMNS_OPTION + " option");
+
+ String encodedColumns = options.get(COLUMNS_OPTION);
+ if (encodedColumns.length() == 0)
+ throw new IllegalArgumentException("The " + COLUMNS_OPTION + " must not be empty");
+
+ combiners = new ColumnSet(Arrays.asList(encodedColumns.split(",")));
}
@Override
public IteratorOptions describeOptions() {
- return new IteratorOptions("comb", "Combiners apply reduce functions to values with identical keys", null, Collections.singletonList("<columnName> null"));
+ return new IteratorOptions(
+ "comb",
+ "Combiners apply reduce functions to values with identical keys",
+ Collections
+ .singletonMap(
+ COLUMNS_OPTION,
+ "<col fam>[:<col qual>]{,<col fam>[:<col qual>]} escape non aplhanum chars using %<hex>."),
+ null);
}
@Override
public boolean validateOptions(Map<String,String> options) {
+ if (!options.containsKey(COLUMNS_OPTION))
+ return false;
+
+ String encodedColumns = options.get(COLUMNS_OPTION);
+ if (encodedColumns.length() == 0)
+ return false;
+
+ for (String columns : encodedColumns.split(",")) {
+ if (!ColumnSet.isValidEncoding(columns))
+ return false;
+ }
+
return true;
}
- /*
- * Adds a column (colf and colq) to an IteratorSetting.
- *
- * @param colf The column family.
- *
- * @param colq The column qualifier (<tt>null</tt> if unspecified).
+ /**
+ * A convenience class for passing column family and column qualifiers
+ */
+ public static class Column extends Pair<Text,Text> {
+
+ public Column(Text columnFamily, Text columnQualifier) {
+ super(columnFamily, columnQualifier);
+ }
+
+ public Column(Text columnFamily) {
+ super(columnFamily, null);
+ }
+
+ public Column(String columnFamily, String columnQualifier) {
+ super(new Text(columnFamily), new Text(columnQualifier));
+ }
+
+ public Column(String columnFamily) {
+ super(new Text(columnFamily), null);
+ }
+
+ }
+
+ /**
+ * A convenience method to set which columns a combiner should be applied to.
*
- * @param is The IteratorSetting to which to add the column parameter.
+ * @param is
+ * iterator settings object to configure
+ * @param columns
+ * a list columns to encode as the value for the combiner column configuration
*/
- public static void addColumn(Text colf, Text colq, IteratorSetting is) {
- String column = PerColumnIteratorConfig.encodeColumns(colf, colq);
- is.addOption(COLUMN_PREFIX + column, "");
+
+ public static void setColumns(IteratorSetting is, List<Column> columns) {
+ String sep = "";
+ StringBuilder sb = new StringBuilder();
+
+ for (Column col : columns) {
+ sb.append(sep);
+ sep = ",";
+ sb.append(ColumnSet.encodeColumns(col.getFirst(), col.getSecond()));
+ }
+
+ is.addOption(COLUMNS_OPTION, sb.toString());
}
}
Modified: incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/ColumnSet.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/ColumnSet.java?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/ColumnSet.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/ColumnSet.java Thu Nov 17 19:39:04 2011
@@ -23,9 +23,9 @@ import java.util.Set;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.iterators.conf.ColumnUtil.ColFamHashKey;
import org.apache.accumulo.core.iterators.conf.ColumnUtil.ColHashKey;
+import org.apache.accumulo.core.util.Pair;
import org.apache.hadoop.io.Text;
-@SuppressWarnings("deprecation")
public class ColumnSet {
private Set<ColFamHashKey> objectsCF;
private Set<ColHashKey> objectsCol;
@@ -42,12 +42,12 @@ public class ColumnSet {
this();
for (String column : objectStrings) {
- PerColumnIteratorConfig pcic = PerColumnIteratorConfig.decodeColumns(column, null);
+ Pair<Text,Text> pcic = ColumnSet.decodeColumns(column);
- if (pcic.getColumnQualifier() == null) {
- add(pcic.getColumnFamily());
+ if (pcic.getSecond() == null) {
+ add(pcic.getFirst());
} else {
- add(pcic.getColumnFamily(), pcic.getColumnQualifier());
+ add(pcic.getFirst(), pcic.getSecond());
}
}
}
@@ -80,4 +80,75 @@ public class ColumnSet {
public boolean isEmpty() {
return objectsCol.size() == 0 && objectsCF.size() == 0;
}
+
+ public static String encodeColumns(Text columnFamily, Text columnQualifier) {
+ StringBuilder sb = new StringBuilder();
+
+ encode(sb, columnFamily);
+ if (columnQualifier != null) {
+ sb.append(':');
+ encode(sb, columnQualifier);
+ }
+
+ return sb.toString();
+ }
+
+ static void encode(StringBuilder sb, Text t) {
+ for (int i = 0; i < t.getLength(); i++) {
+ int b = (0xff & t.getBytes()[i]);
+
+ // very inefficient code
+ if ((b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z') || (b >= '0' && b <= '9') || b == '_' || b == '-') {
+ sb.append((char) b);
+ } else {
+ sb.append('%');
+ sb.append(String.format("%02x", b));
+ }
+ }
+ }
+
+ public static boolean isValidEncoding(String enc) {
+ for (char c : enc.toCharArray()) {
+ boolean validChar = (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' || c == '-' || c == ':' || c == '%';
+ if (!validChar)
+ return false;
+ }
+
+ return true;
+ }
+
+ public static Pair<Text,Text> decodeColumns(String columns) {
+ if (!isValidEncoding(columns))
+ throw new IllegalArgumentException("Invalid encoding " + columns);
+
+ String[] cols = columns.split(":");
+
+ if (cols.length == 1) {
+ return new Pair<Text,Text>(decode(cols[0]), null);
+ } else if (cols.length == 2) {
+ return new Pair<Text,Text>(decode(cols[0]), decode(cols[1]));
+ } else {
+ throw new IllegalArgumentException(columns);
+ }
+ }
+
+ static Text decode(String s) {
+ Text t = new Text();
+
+ byte[] sb = s.getBytes();
+
+ // very inefficient code
+ for (int i = 0; i < sb.length; i++) {
+ if (sb[i] != '%') {
+ t.append(new byte[] {sb[i]}, 0, 1);
+ } else {
+ byte hex[] = new byte[] {sb[++i], sb[++i]};
+ String hs = new String(hex);
+ int b = Integer.parseInt(hs, 16);
+ t.append(new byte[] {(byte) b}, 0, 1);
+ }
+ }
+
+ return t;
+ }
}
Modified: incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/PerColumnIteratorConfig.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/PerColumnIteratorConfig.java?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/PerColumnIteratorConfig.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/iterators/conf/PerColumnIteratorConfig.java Thu Nov 17 19:39:04 2011
@@ -58,64 +58,22 @@ public class PerColumnIteratorConfig {
}
private static String encodeColumns(PerColumnIteratorConfig pcic) {
- return encodeColumns(pcic.colf, pcic.colq);
+ return ColumnSet.encodeColumns(pcic.colf, pcic.colq);
}
public static String encodeColumns(Text columnFamily, Text columnQualifier) {
- StringBuilder sb = new StringBuilder();
-
- encode(sb, columnFamily);
- if (columnQualifier != null) {
- sb.append(':');
- encode(sb, columnQualifier);
- }
-
- return sb.toString();
+ return ColumnSet.encodeColumns(columnFamily, columnQualifier);
}
-
- private static void encode(StringBuilder sb, Text t) {
- for (int i = 0; i < t.getLength(); i++) {
- int b = (0xff & t.getBytes()[i]);
-
- // very inefficient code
- if ((b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z') || (b >= '0' && b <= '9') || b == '_' || b == '-') {
- sb.append((char) b);
- } else {
- sb.append('%');
- sb.append(String.format("%02x", b));
- }
- }
- }
-
+
public static PerColumnIteratorConfig decodeColumns(String columns, String className) {
String[] cols = columns.split(":");
if (cols.length == 1) {
- return new PerColumnIteratorConfig(decode(cols[0]), className);
+ return new PerColumnIteratorConfig(ColumnSet.decode(cols[0]), className);
} else if (cols.length == 2) {
- return new PerColumnIteratorConfig(decode(cols[0]), decode(cols[1]), className);
+ return new PerColumnIteratorConfig(ColumnSet.decode(cols[0]), ColumnSet.decode(cols[1]), className);
} else {
throw new IllegalArgumentException(columns);
}
}
-
- private static Text decode(String s) {
- Text t = new Text();
-
- byte[] sb = s.getBytes();
-
- // very inefficient code
- for (int i = 0; i < sb.length; i++) {
- if (sb[i] != '%') {
- t.append(new byte[] {sb[i]}, 0, 1);
- } else {
- byte hex[] = new byte[] {sb[++i], sb[++i]};
- String hs = new String(hex);
- int b = Integer.parseInt(hs, 16);
- t.append(new byte[] {(byte) b}, 0, 1);
- }
- }
-
- return t;
- }
}
Modified: incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/util/shell/commands/SetIterCommand.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/util/shell/commands/SetIterCommand.java?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/util/shell/commands/SetIterCommand.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/util/shell/commands/SetIterCommand.java Thu Nov 17 19:39:04 2011
@@ -29,18 +29,18 @@ import org.apache.accumulo.core.client.A
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.iterators.AggregatingIterator;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.NoLabelIterator;
import org.apache.accumulo.core.iterators.OptionDescriber;
+import org.apache.accumulo.core.iterators.OptionDescriber.IteratorOptions;
import org.apache.accumulo.core.iterators.RegExIterator;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.VersioningIterator;
-import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
-import org.apache.accumulo.core.iterators.OptionDescriber.IteratorOptions;
import org.apache.accumulo.core.iterators.aggregation.Aggregator;
import org.apache.accumulo.core.iterators.user.AgeOffFilter;
import org.apache.accumulo.core.util.shell.Shell;
-import org.apache.accumulo.core.util.shell.ShellCommandException;
import org.apache.accumulo.core.util.shell.Shell.Command;
+import org.apache.accumulo.core.util.shell.ShellCommandException;
import org.apache.accumulo.core.util.shell.ShellCommandException.ErrorCode;
import org.apache.accumulo.start.classloader.AccumuloClassLoader;
import org.apache.commons.cli.CommandLine;
@@ -139,6 +139,10 @@ public class SetIterCommand extends Comm
if (itopts.getName() == null)
throw new IllegalArgumentException(className + " described its default distinguishing name as null");
+ String shortClassName = className;
+ if (className.contains("."))
+ shortClassName = className.substring(className.lastIndexOf('.') + 1);
+
Map<String,String> localOptions = new HashMap<String,String>();
do {
// clean up the overall options that caused things to fail
@@ -152,7 +156,7 @@ public class SetIterCommand extends Comm
String prompt;
if (itopts.getNamedOptions() != null) {
for (Entry<String,String> e : itopts.getNamedOptions().entrySet()) {
- prompt = Shell.repeat("-", 10) + "> set " + className + " parameter " + e.getKey() + ", " + e.getValue() + ": ";
+ prompt = Shell.repeat("-", 10) + "> set " + shortClassName + " parameter " + e.getKey() + ", " + e.getValue() + ": ";
input = reader.readLine(prompt);
if (input == null) {
@@ -171,7 +175,7 @@ public class SetIterCommand extends Comm
reader.printString(Shell.repeat("-", 10) + "> entering options: " + desc + "\n");
input = "start";
while (true) {
- prompt = Shell.repeat("-", 10) + "> set " + className + " option (<name> <value>, hit enter to skip): ";
+ prompt = Shell.repeat("-", 10) + "> set " + shortClassName + " option (<name> <value>, hit enter to skip): ";
input = reader.readLine(prompt);
if (input == null) {
Modified: incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java Thu Nov 17 19:39:04 2011
@@ -84,7 +84,7 @@ public class MockConnectorTest {
c.tableOperations().create(table);
Class<? extends SortedKeyValueIterator<Key,Value>> clazz = SummingCombiner.class;
IteratorSetting is = new IteratorSetting(10, "String Summation", clazz);
- Combiner.addColumn(new Text("day"), null, is);
+ Combiner.setColumns(is, Collections.singletonList(new Combiner.Column("day")));
is.addOption(SummingCombiner.TYPE, SummingCombiner.Type.STRING.name());
c.tableOperations().attachIterator(table, is);
String keys[][] = { {"foo", "day", "20080101"}, {"foo", "day", "20080101"}, {"foo", "day", "20080103"}, {"bar", "day", "20080101"},
Modified: incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTest.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTest.java?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTest.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTest.java Thu Nov 17 19:39:04 2011
@@ -94,6 +94,7 @@ public class CombinerTest {
Map<String,String> opts = new HashMap<String,String>();
opts.put(SummingCombiner.TYPE, SummingCombiner.Type.VARNUM.name());
+ opts.put(Combiner.COLUMNS_OPTION, "2");
ai.init(new SortedMapIterator(tm1), opts, null);
ai.seek(new Range(), EMPTY_COL_FAMS, false);
@@ -158,7 +159,7 @@ public class CombinerTest {
Map<String,String> opts = new HashMap<String,String>();
- opts.put(Combiner.COLUMN_PREFIX + "cf001", null);
+ opts.put(Combiner.COLUMNS_OPTION, "cf001");
opts.put(SummingCombiner.TYPE, SummingCombiner.Type.VARNUM.name());
ai.init(new SortedMapIterator(tm1), opts, null);
@@ -225,7 +226,7 @@ public class CombinerTest {
Map<String,String> opts = new HashMap<String,String>();
- opts.put(Combiner.COLUMN_PREFIX + "cf001", "null");
+ opts.put(Combiner.COLUMNS_OPTION, "cf001");
opts.put(SummingCombiner.TYPE, SummingCombiner.Type.LONG.name());
ai.init(new SortedMapIterator(tm1), opts, null);
@@ -295,7 +296,7 @@ public class CombinerTest {
Map<String,String> opts = new HashMap<String,String>();
- opts.put(Combiner.COLUMN_PREFIX + "cf001", null);
+ opts.put(Combiner.COLUMNS_OPTION, "cf001");
opts.put(SummingCombiner.TYPE, SummingCombiner.Type.STRING.name());
ai.init(new SortedMapIterator(tm1), opts, null);
@@ -366,7 +367,7 @@ public class CombinerTest {
Combiner ai = new SummingCombiner();
Map<String,String> opts = new HashMap<String,String>();
- opts.put(Combiner.COLUMN_PREFIX + "cf001", null);
+ opts.put(Combiner.COLUMNS_OPTION, "cf001");
opts.put(SummingCombiner.TYPE, SummingCombiner.Type.STRING.name());
List<SortedKeyValueIterator<Key,Value>> sources = new ArrayList<SortedKeyValueIterator<Key,Value>>(3);
@@ -397,7 +398,7 @@ public class CombinerTest {
Map<String,String> opts = new HashMap<String,String>();
- opts.put(Combiner.COLUMN_PREFIX + "cf001", null);
+ opts.put(Combiner.COLUMNS_OPTION, "cf001");
opts.put(SummingCombiner.TYPE, SummingCombiner.Type.VARNUM.name());
ai.init(new SortedMapIterator(tm1), opts, new DefaultIteratorEnvironment());
@@ -426,7 +427,7 @@ public class CombinerTest {
Map<String,String> opts = new HashMap<String,String>();
- opts.put(Combiner.COLUMN_PREFIX + "cf001", "null");
+ opts.put(Combiner.COLUMNS_OPTION, "cf001");
opts.put(SummingCombiner.TYPE, SummingCombiner.Type.LONG.name());
ai.init(new SortedMapIterator(tm1), opts, new DefaultIteratorEnvironment());
@@ -488,7 +489,7 @@ public class CombinerTest {
Map<String,String> opts = new HashMap<String,String>();
- opts.put(Combiner.COLUMN_PREFIX + "cf001", null);
+ opts.put(Combiner.COLUMNS_OPTION, "cf001");
opts.put(SummingCombiner.TYPE, SummingCombiner.Type.VARNUM.name());
ai.init(new SortedMapIterator(tm1), opts, null);
@@ -542,7 +543,7 @@ public class CombinerTest {
Map<String,String> opts = new HashMap<String,String>();
- opts.put(Combiner.COLUMN_PREFIX + "cf001", null);
+ opts.put(Combiner.COLUMNS_OPTION, "cf001");
opts.put(SummingCombiner.TYPE, type);
ai.init(new SortedMapIterator(tm1), opts, null);
Added: incubator/accumulo/branches/1.4/src/examples/src/main/java/org/apache/accumulo/examples/combiner/StatsCombiner.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/examples/src/main/java/org/apache/accumulo/examples/combiner/StatsCombiner.java?rev=1203345&view=auto
==============================================================================
--- incubator/accumulo/branches/1.4/src/examples/src/main/java/org/apache/accumulo/examples/combiner/StatsCombiner.java (added)
+++ incubator/accumulo/branches/1.4/src/examples/src/main/java/org/apache/accumulo/examples/combiner/StatsCombiner.java Thu Nov 17 19:39:04 2011
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.examples.combiner;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.Combiner;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+
+/**
+ *
+ */
+public class StatsCombiner extends Combiner {
+
+ public static final String RADIX_OPTION = "radix";
+
+ private int radix = 10;
+
+ @Override
+ public Value reduce(Key key, Iterator<Value> iter) {
+
+ long min = Long.MAX_VALUE;
+ long max = Long.MIN_VALUE;
+ long sum = 0;
+ long count = 0;
+
+ while (iter.hasNext()) {
+ String stats[] = iter.next().toString().split(",");
+
+ if (stats.length == 1) {
+ long val = Long.parseLong(stats[0], radix);
+ min = Math.min(val, min);
+ max = Math.max(val, max);
+ sum += val;
+ count += 1;
+ } else {
+ min = Math.min(Long.parseLong(stats[0], radix), min);
+ max = Math.max(Long.parseLong(stats[1], radix), max);
+ sum += Long.parseLong(stats[2], radix);
+ count += Long.parseLong(stats[3], radix);
+ }
+ }
+
+ String ret = Long.toString(min, radix) + "," + Long.toString(max, radix) + "," + Long.toString(sum, radix) + "," + Long.toString(count, radix);
+ return new Value(ret.getBytes());
+ }
+
+ @Override
+ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
+ super.init(source, options, env);
+
+ if (options.containsKey(RADIX_OPTION))
+ radix = Integer.parseInt(options.get(RADIX_OPTION));
+ else
+ radix = 10;
+ }
+
+ @Override
+ public IteratorOptions describeOptions() {
+ IteratorOptions io = super.describeOptions();
+ io.setName("statsCombiner");
+ io.setDescription("Combiner that keeps track of min, max, sum, and count");
+ io.addNamedOption(RADIX_OPTION, "radix/base of the numbers");
+ return io;
+ }
+
+ @Override
+ public boolean validateOptions(Map<String,String> options) {
+ if (!super.validateOptions(options))
+ return false;
+
+ if (options.containsKey(RADIX_OPTION) && !options.get(RADIX_OPTION).matches("\\d+"))
+ return false;
+
+ return true;
+ }
+
+ /**
+ * A convenience method for setting the expected base/radix of the numbers
+ *
+ * @param iterConfig
+ * Iterator settings to configure
+ * @param maxSize
+ * Maximum set size
+ */
+ public static void setRadix(IteratorSetting iterConfig, int base) {
+ iterConfig.addOption(RADIX_OPTION, base + "");
+ }
+}
Modified: incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/test/functional/ServerSideErrorTest.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/test/functional/ServerSideErrorTest.java?rev=1203345&r1=1203344&r2=1203345&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/test/functional/ServerSideErrorTest.java (original)
+++ incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/test/functional/ServerSideErrorTest.java Thu Nov 17 19:39:04 2011
@@ -58,7 +58,7 @@ public class ServerSideErrorTest extends
getConnector().tableOperations().create("tt");
IteratorSetting is = new IteratorSetting(5, "Bad Aggregator", BadCombiner.class);
- Combiner.addColumn(new Text("acf"), null, is);
+ Combiner.setColumns(is, Collections.singletonList(new Combiner.Column("acf")));
getConnector().tableOperations().attachIterator("tt", is);
BatchWriter bw = getConnector().createBatchWriter("tt", 1000000, 60000l, 2);