Built-in filters

Some useful filters are provided with Accumulo:

AgeOffFilter

Removes keys when their timestamps differ from the current time by more than a specified parameter (in milliseconds).

ColumnAgeOffFilter

Stores a separate age-off parameter for each column, to age off columns at different rates.

TimestampFilter

Only keeps keys with timestamps earlier and/or later than given start and end parameters.

RegExFilter

Returns key-value pairs that match a Java regular expression in a particular portion of the key or value (the row, column family, column qualifier, or value). Regular expressions can be provided for any subset of these four, and matches can be determined by ORing or ANDing together the results of each individual regular expression.

GrepIterator

An iterator that matches exact strings to all key-value pairs scanned. This is great for doing one-time scans of a table. If you find yourself using this iterator frequently you might want to look into secondary indexes, as described in “Secondary Indexing”.

ReqVisFilter

Removes keys with empty column visibilities.

LargeRowFilter

Suppresses entire rows that have more than a configurable number of columns. It buffers the row in memory when determining whether or not it should be suppressed, so the specified number of columns should not be too large.

RowFilter

An abstract iterator that decides whether or not to include an entire row. Subclasses of RowFilter must implement an acceptRow() method that takes as a parameter a SortedKeyValueIterator<Key,Value> (which will be limited to the row being decided upon) and returns a Boolean. This allows you to decide to include a row based on several features of the row, such as the presence of two or more columns or a relationship between values.

All filters can be configured to reverse their logic by using the setNegate() method:

IteratorSetting setting;
boolean negate = true;

MyFilter.setNegate(setting, negate);

Custom filters

Custom filters can be written by extending org.apache.accumulo.core.iterators.Filter. Subclasses of Filter must implement an accept() method that takes Key and Value as parameters and returns a boolean.

The interface is as follows:

public abstract class Filter extends WrappingIterator
    implements OptionDescriber {
  ...
  /**
   * @return <tt>true</tt> if the key/value pair is accepted by the filter.
   */
  public abstract boolean accept(Key k, Value v);

}

We’ll walk through an example next to see how to make our own custom filter.

Custom filtering iterator example

Here we’ll create a custom filtering iterator that returns key-value pairs only if the value, interpreted as a number, is greater than a user-provided threshold.

First we create our filter class by extending Filter and defining our accept() method:

public class GreaterThanFilterExample extends Filter {

  private static final String GREATER_THAN_CRITERION = "greaterThanOption";

  private long threshold = 0;

  @Override
  public boolean accept(Key k, Value v) {
    try {
      long num = Long.parseLong(new String(v.get()));
      return num > threshold;
    } catch(NumberFormatException ex) {
      // continue and return false
    }

    return false;
  }
  ...
}

We need to write a few other methods to make users aware of the required threshold setting:

@Override
public IteratorOptions describeOptions() {
  IteratorOptions opts = super.describeOptions();
  opts.addNamedOption(GREATER_THAN_CRITERION,
      "Only return values greater than given numerical value");
  return opts;
}

@Override
public boolean validateOptions(Map<String,String> options) {
  if(!super.validateOptions(options) ||
      !options.containsKey(GREATER_THAN_CRITERION)) {
    return false;
  }

  String gtString = options.get(GREATER_THAN_CRITERION);
  try {
    Long.parseLong(gtString);
  }
  catch (NumberFormatException e) {
    return false;
  }

  return true;
}

Also, the convention is for iterators to provide static methods for filling out options on IteratorSetting objects. We’ll add one for our threshold:

public static void setThreshold(
        final IteratorSetting setting,
        final int threshold) {

  setting.addOption(GREATER_THAN_CRITERION, Integer.toString(threshold));
}

Next we want to fetch the threshold from the iterator options whenever our iterator is used. Iterator classes are set up and torn down at the discretion of the tablet server. The init() method will allow us to perform some setup before the accept() method is called:

@Override
public void init(SortedKeyValueIterator<Key,Value> source,
    Map<String,String> options, IteratorEnvironment env) throws IOException {
  super.init(source, options, env);
  if (options.containsKey(GREATER_THAN_CRITERION)) {
    String gtString = options.get(GREATER_THAN_CRITERION);

    threshold = Long.parseLong(gtString);
  }
}

Finally, we can test our filter. We’ll use the ExampleMiniCluster to create a test table and apply our iterator to it:

Random random = new Random();

Connector conn = ExampleMiniCluster.getConnector();
TableOperations ops = conn.tableOperations();

ops.create("testTable");

// insert some data
BatchWriter writer = conn.createBatchWriter("testTable",
    new BatchWriterConfig());

for(int i=0; i < 30; i++) {

  int rowNum = random.nextInt(100);
  int colNum = random.nextInt(100);
  int value = random.nextInt(100);

  Mutation m = new Mutation("row" + rowNum);

  m.put("", "col" + colNum, Integer.toString(value));
  writer.addMutation(m);
}

writer.flush();

Now that our test table is full of values with random numbers between 0 and 100, we’ll add our iterator and scan:

IteratorSetting setting = new IteratorSetting(15, "gtf",
    GreaterThanFilterExample.class.getName());
GreaterThanFilterExample.setThreshold(setting, 80);

conn.tableOperations().attachIterator("testTable", setting);

// we could, instead, set our iterator just for this scan
//scanner.addScanIterator(setting);

// check for the existence of our iterator
for(Map.Entry<String, EnumSet<IteratorUtil.IteratorScope>> i :
    conn.tableOperations().listIterators("testTable").entrySet()) {
  System.out.println(i);
}

Checking for our iterator on the table shows:

gtf=[majc, minc, scan]
vers=[majc, minc, scan]

The only thing left to do is to scan our table and check the output:

// scan whole table 
Scanner scanner = conn.createScanner("testTable", Authorizations.EMPTY);
for(Map.Entry<Key, Value> e : scanner) {
  System.out.println(e);
}

Note that using an iterator to scan an entire table is not scalable to large tables, but is good enough for our test here.

The output we get should show no values under our threshold, 80:

row0 :col1 [] 1409452474218 false 95
row13 :col94 [] 1409452474218 false 91
row32 :col70 [] 1409452474218 false 84
row34 :col56 [] 1409452474218 false 92
row49 :col94 [] 1409452474218 false 95
row59 :col86 [] 1409452474218 false 92
row93 :col60 [] 1409452474218 false 91

To deploy this iterator to a production cluster, we’ll need to build a JAR that contains our class and place it where server processes can load it. See “Deploying JARs” for details.

Combiners for incrementing or appending updates

Combiners can help when an application calls for doing updates in which new values should be appended or otherwise combined with existing values. In many systems this requires reading the existing values first, applying the combination logic, and writing back the combined value. Accumulo can combine updates very efficiently by allowing multiple partially combined values to coexist until the finalized answer is needed—for example when a value is read by a client performing a scan. This allows these types of updates to be applied to a table with the same performance as simple inserts or overwriting updates.

For example, if we would like to update a numerical value by adding a new amount to the existing value, we don’t have to somehow lock the row and column, read out the old value, add our new amount to it, and write it back. We can simply insert the amount to be added and instruct the server to add up all the existing values for that row and column. When the server writes values to disk it always writes the combined value to cut down on the partial values that are stored.

To illustrate the difference, consider the following scenario. Let’s say we are maintaining a summary of the number of times we have seen each word in a corpus. When we would like to update results we have two choices. One approach is to have the application read the old value, add the new value, and write the combined value back (Figure 6-2).

Figure 6-2. Update by performing a read then an overwrite

Another, much faster approach is to do a simple insert of the additional counts, and let a combiner do the final summation (Figure 6-3).

Figure 6-3. Update by performing an insert and letting combiner do final summation

Built-in combiners

Accumulo comes with a number of combiners. These are in the org.apache.accumulo.core.iterators.user package:

LongCombiner

An abstract combiner that interprets Accumulo Values as Java Long objects. It comes with three possible encoding types: STRING, which prints and parses the number as a string; LONG, which encodes the number in exactly 8 binary bytes; and VARNUM, which uses a variable-length binary encoding.

MaxCombiner

Extends the LongCombiner, interpreting values as Longs, and returns the maximum Long for each set of values.

MinCombiner

Extends the LongCombiner, interpreting values as Longs, and returns the minimum Long for each set of values.

SummingCombiner

Extends the LongCombiner, interpreting values as Longs, and returns the sum of the set of values.

SummingArrayCombiner

Interprets Values as an array of Longs, and returns an array of sums. If the arrays are not the same length, the shorter arrays are padded with zeros to equal the length of the largest array.

BigDecimalCombiner

An abstract combiner that interprets Accumulo values as BigDecimals.

An additional example combiner is the StatsCombiner in the org.apache.accumulo.examples.simple.combiner package. Use of this combiner is illustrated in Accumulo’s README.combiner example.

We saw an example of setting up the SummingCombiner via the API in “Iterator Configuration Example”.

Next we’ll look at writing our own combiner.

Custom combiners

Custom combiners can be written by extending org.apache.accumulo.core.iterators.Combiner. Subclasses of Combiner must implement a reduce() method that takes Key and Iterator<Value> as parameters and returns a Value:

public abstract Value reduce(Key key, Iterator<Value> iter);

If the values are always interpreted as a particular Java type, the TypedValueCombiner<V> can be used. This combiner uses an Encoder to translate the type V to and from a byte array. Subclasses of TypedValueCombiner implement a typedReduce() method that takes Key and Iterator<V> as parameters and returns an object of type V:

public abstract class TypedValueCombiner<V> extends Combiner {
  ...
  public abstract V typedReduce(Key key, Iterator<V> iter);

}

A combiner that extends TypedValueCombiner should also have a class implementing the Encoder interface for converting values to byte arrays and back:

public abstract class TypedValueCombiner<V> extends Combiner {
  ...
  public interface Encoder<V> {
    byte[] encode(V v);

    V decode(byte[] b) throws ValueFormatException;
  }
  ...
}

Custom combiner example

We’ll implement a combiner that keeps track of the number of items seen and the sum of the items seen in order to produce an average. We can’t simply store the average because we’ll lose information about the number of items seen so far and won’t know how much relative weight to apply to new items.

First, we’ll implement an Encoder to store a Long and a Double, for the number of items seen, and the partial total:

public class RunningAverageCombiner
    extends TypedValueCombiner<Pair<Long,Double>> {

  ...
  public static class LongDoublePairEncoder
      implements Encoder<Pair<Long,Double>> {

    @Override
    public byte[] encode(Pair<Long, Double> v) {
      String s = Long.toString(v.getFirst()) + ":" +
        Double.toString(v.getSecond());

      return s.getBytes(StandardCharsets.UTF_8);
    }

    @Override
    public Pair<Long, Double> decode(byte[] b) throws ValueFormatException {
      String s = new String(b, StandardCharsets.UTF_8);
      String[] parts = s.split(":");
      return new Pair<>(Long.parseLong(parts[0]), Double.parseDouble(parts[1]));
    }
  }
  ...
}

This class will allow us to write string values representing a count and a running total to be averaged.

Now we’ll define our reduce() function, which will tell tablet servers how to combine multiple versions of values for a key. In our case, we simply keep track of the number of items we’ve seen, and a running total. At any given time, a client can divide the running total by the number of items seen to get the current average:

  ...
  @Override
  public void init(SortedKeyValueIterator<Key,Value> source,
      Map<String,String> options, IteratorEnvironment env) throws IOException {
    super.init(source, options, env);
    setEncoder(new LongDoublePairEncoder()); 
  }

  @Override
  public Pair<Long,Double> typedReduce(Key key,
      Iterator<Pair<Long,Double>> iter) {

    Long count = 0L;
    Double sum = 0.0;

    while(iter.hasNext()) {
      Pair<Long,Double> pair = iter.next();

      count += pair.getFirst();
      sum += pair.getSecond();
    }

    return new Pair<>(count, sum);
  }
  ...

Be sure to initialize the encoder to be used here.

Now we can test our combiner on a table. First we’ll create a table, remove the VersioningIterator so that our combiner receives all versions of each key, and set our custom combiner:

Connector conn = ExampleMiniCluster.getConnector();

TableOperations ops = conn.tableOperations();
ops.create("testTable");

// remove versioning iterator
ops.removeIterator("testTable", "vers", EnumSet.allOf(IteratorScope.class));

// configure our iterator
IteratorSetting setting = new IteratorSetting(10, "rac",
    RunningAverageCombiner.class); 
RunningAverageCombiner.setCombineAllColumns(setting, true); 
RunningAverageCombiner.setLossyness(setting, false); 

// attach to table for all scopes
ops.attachIterator("testTable", setting);

Give our combiner a priority of 10, name it rac, and provide the class.

Set our combiner to operate on all columns.

Instruct the combiner to throw exceptions when values fail to decode with the given encoder, instead of silently discarding those values.

Now we can insert some test data. We’ll generate random numbers in a particular range and insert them as multiple versions of the same key. This means the row ID, column family, column qualifier, and column visibility must be the same. Values in different keys will not be combined together, only values representing multiple versions of the same key.

We’ll generate numbers uniformly at random between 4.5 and 6.5. Our average should be somewhere around 5.5:

BatchWriter writer = conn.createBatchWriter("testTable",
    new BatchWriterConfig());

// begin writing numbers to our table
Random random = new Random();

for(int i = 0; i < 5; i++) {
  Mutation m = new Mutation("heights");
  m.put("", "average", "1:" + (random.nextDouble() * 2 + 4.5));
  writer.addMutation(m);
}
writer.flush();

Now we’ll perform a scan on our table. This will cause our combiner to examine the key-value pairs we’ve written and give it a chance to combine them before returning any to our client.

We’ll use the same LongDoublePairEncoder to decode values into our count and running total. Then we’ll divide the total by the count and print out the average:

LongDoublePairEncoder enc = new LongDoublePairEncoder();

Scanner scanner = conn.createScanner("testTable", Authorizations.EMPTY);
for(Map.Entry<Key, Value> e : scanner) {
  Pair<Long,Double> pair = enc.decode(e.getValue().get());

  double average = pair.getSecond() / pair.getFirst();
  System.out.println(average);
}

Our output consists of one number:

5.975622642042859

Next we write one hundred more versions of our key. After scanning again our average is a little closer to 5.5:

5.329055520715937

To load this combiner into tablet servers in a production cluster we’ll need to package our combiner class and encoder class into a JAR file and deploy it to all the servers. See “Deploying JARs” for details. Be sure to compile your combiner against the same version of Accumulo that is running on your cluster.

WholeRowIterator example

When a table is scanned, key-value pairs are streamed back to the client in sorted order, one after another. Sets of key-value pairs with the same row ID are considered to be the same row. Clients must examine the row ID to determine which key-value pairs belong in which row, unless they are grouped using the RowIterator described in “Grouping by Rows”.

If all the key-value pairs in a row can fit in memory comfortably, we can choose to use the WholeRowIterator to get a set of key-value pairs for one row grouped together in a convenient data structure.

For this example, we’ll create a test table with 100 rows containing 100 columns each. Each row will be read into client memory completely, and we can decode it into separate columns and access them in any order we choose:

Connector conn = ExampleMiniCluster.getConnector();

TableOperations ops = conn.tableOperations();
ops.create("testTable");

BatchWriter writer = conn.createBatchWriter("testTable",
    new BatchWriterConfig());

for(int i=1; i <= 100; i++) {
  Mutation m = new Mutation("row" + String.format("%02d", i));

  for(int j = 1; j <= 100; j++) {
    m.put("", "col" + j, Integer.toString(i * j));
  }

  writer.addMutation(m);
}
writer.flush();

Next we’ll create a scanner and add the WholeRowIterator to it:

Scanner scanner = conn.createScanner("testTable", Authorizations.EMPTY);
scanner.setRange(new Range("row50", "row60"));

IteratorSetting setting = new IteratorSetting(30, "wri", WholeRowIterator.class);

scanner.addScanIterator(setting);

The WholeRowIterator has a decode() method that will unpack one row’s worth of data for us into a SortedMap<Key,Value> object. We’ll use some convenience methods for grabbing the row ID and creating a map of just column names to values:

private static byte[] getRow(SortedMap<Key,Value> row) {
  return row.entrySet().iterator().next().getKey().getRow().getBytes();
}

private static SortedMap<String,Value> columnMap(SortedMap<Key,Value> row) {

  TreeMap<String,Value> colMap = new TreeMap<>();
  for(Map.Entry<Key, Value> e : row.entrySet()) {

    String cf = e.getKey().getColumnFamily().toString();
    String cq = e.getKey().getColumnQualifier().toString();

    colMap.put(cf + ":" + cq, e.getValue());
  }
  return colMap;
}

Now we can perform our scan and operate on one row’s worth of data at a time, grabbing columns in whatever order we want:

Scanner scanner = conn.createScanner("testTable", Authorizations.EMPTY);
scanner.setRange(new Range("row50", "row60"));
// we could choose a subset of columns via fetchColumn() too

IteratorSetting setting = new IteratorSetting(30, "wri", WholeRowIterator.class);
scanner.addScanIterator(setting);

for(Map.Entry<Key, Value> e : scanner) {
  SortedMap<Key, Value> rowData =
    WholeRowIterator.decodeRow(e.getKey(), e.getValue()); 

  byte[] row = getRow(rowData);
  SortedMap<String, Value> columns = columnMap(rowData);

  System.out.println("
row	" + new String(row));
  System.out.println(":col31	" + columns.get(":col31"));
  System.out.println(":col15	" + columns.get(":col15"));
}

Decode the key-value pair to get the row’s columns.

Running our code produces the following output, showing the out-of-order access of columns within each row:

row  row50
:col31 1550
:col15 750

row  row51
:col31 1581
:col15 765

...

row  row60
:col31 1860
:col15 900

Low-level iterator API

Filtering iterators and combiners are special cases of general iterators. There is a low-level iterator API that can be used to create new types of iterators. It is much more complicated than the API for implementing new filters and combiners, however.

Lower-level iterators at the very least implement the SortedKeyValueIterator<Key,Value> interface:

public interface SortedKeyValueIterator<K extends WritableComparable<?>,
    V extends Writable> {

  void init(SortedKeyValueIterator<K,V> source, Map<String,String> options,
      IteratorEnvironment env) throws IOException;

  boolean hasTop();

  void next() throws IOException;

  void seek(Range range, Collection<ByteSequence> columnFamilies,
      boolean inclusive) throws IOException;

  K getTopKey();

  V getTopValue();

  SortedKeyValueIterator<K,V> deepCopy(IteratorEnvironment env);
}

Some iterators choose to extend the WrappingIterator or SkippingIterator classes.

We’ll implement a simple iterator that takes advantage of one of the features of the lower-level API: the ability to seek ahead to a new key-value pair. Our iterator will simply return the first column and its value for each row we scan. After reading the first column for a row, it will seek to the next row.

Accumulo already has an implementation of an iterator that performs this same function and includes some additional optimization: the FirstEntryInRowIterator.

The code for our iterator will begin at the WrappingIterator, rather than simply implementing SortedKeyValueIterator. This will help guarantee that we don’t call methods out of order.

We’ll first create our class:

public class FirstColumnIterator extends WrappingIterator  {

  private Range range;
  private boolean inclusive;
  private Collection<ByteSequence> columnFamilies;
  private boolean done;

  public FirstColumnIterator() {} 

  public FirstColumnIterator(FirstColumnIterator aThis,
      IteratorEnvironment env) { 
    super();
    setSource(aThis.getSource().deepCopy(env));
  }

  @Override
  public void init(SortedKeyValueIterator<Key,Value> source,
      Map<String,String> options, IteratorEnvironment env) throws IOException {
    super.init(source, options, env);
  }
  ...

We need a public default constructor because this is the constructer Accumulo will use to instantiate the iterator.

Additional constructors are optional but may be helpful in implementing the deepCopy method.

Next we’ll implement just the seek(), next(), hasTop(), and deepCopy() methods. seek() is called first by the tablet server, and we’ll use it to store off some variables we’ll need to do our seeking later. It will also handle the case in which a seek is made to the middle of a row. next() is called to prepare the next key-value pair for retrieval by getTopKey() and getTopValue(), if the key-value pair exists. hasTop() indicates whether a key-value pair exists for retrieval.

We call on our superclass, the WrappingIterator, to seek to the next row in our source iterator. Every iterator has a source that ultimately goes back to files and in-memory data structures holding key-value pairs. Our iterator simply advances the source iterator past all but the first column of each row and notes when the end of the range is reached:

@Override
public void next() throws IOException {
  if(done) {
    return;
  }

  // create a new range to seek to
  Key nextKey = getSource().getTopKey().followingKey(PartialKey.ROW); 
  if(range.afterEndKey(nextKey)) { 
    done = true;
  }
  else {
    Range nextRange = new Range(nextKey, true, range.getEndKey(),
        range.isEndKeyInclusive()); 
    getSource().seek(nextRange, columnFamilies, inclusive); 
  }
}

@Override
public boolean hasTop() {
  return !done && getSource().hasTop(); 
}

@Override
public void seek(Range range, Collection<ByteSequence> columnFamilies,
    boolean inclusive) throws IOException {
  this.range = range;
  this.columnFamilies = columnFamilies; 
  this.inclusive = inclusive;

  done = false;

  Key startKey = range.getStartKey();
  Range seekRange = new Range(
      startKey == null ? null : new Key(startKey.getRow()), true,
      range.getEndKey(), range.isEndKeyInclusive());
  super.seek(seekRange, columnFamilies, inclusive); 

  if (getSource().hasTop()) {
    if (range.beforeStartKey(getSource().getTopKey())) 
      next();
  }
}

@Override
public SortedKeyValueIterator<Key,Value> deepCopy(IteratorEnvironment env) {
  return new FirstColumnIterator(this, env);
}

This method will get a key that has the first possible row ID after the current key.

Check to see if we’ve reached the end of the range requested by the user and return if so.

Create a range almost exactly like the one that was first given to us by the seek() method.

Advance the source iterator past all the columns of this row to the first column of the next row.

Return false if next() determined the end of the range has been reached, or if there is no more data in the source iterator.

Save these seek parameters off for doing seeks in our next method.

Construct a range that starts at the beginning of the row containing our start key, and seek our superclass, which will seek the source iterator.

Skip to the next row if the top key is before our start key, meaning that we received a seek to the middle of a row.

Now we can apply our iterator to data in a table. First, we’ll create a simple table of 100 rows with 100 columns each. This will make it easy to see if our iterator is working:

Connector conn = ExampleMiniCluster.getConnector();

TableOperations ops = conn.tableOperations();
ops.create("testTable");

BatchWriter writer = conn.createBatchWriter("testTable",
    new BatchWriterConfig());

for(int i=0; i < 100; i++) {
  Mutation m = new Mutation("row" + String.format("%02d", i));

  for(int j = 0; j < 100; j++) {
    m.put("", String.format("col%02d", j), i + " " + j);
  }

  writer.addMutation(m);
}
writer.flush();

Now we’ll scan the table and just count the key-value pairs without our iterator so we’ll have something to compare our iterator results to:

Scanner scanner = conn.createScanner("testTable", Authorizations.EMPTY);

// count items returned
int returned = 0;
for(Map.Entry<Key, Value> e : scanner)
  returned++;

System.out.println("items returned: " + returned);

Our output shows:

items returned: 10000

Now we’ll apply our iterator and see if we only see the first column for every row:

IteratorSetting setting = new IteratorSetting(30, "fci",
    FirstColumnIterator.class);
scanner.addScanIterator(setting);

returned = 0;
for(Map.Entry<Key, Value> e : scanner) {
  System.out.println(e);
  returned++;
}

System.out.println("items returned: " + returned);

Now our output shows the rows and columns returned, with only the first column of each row, and a total count of 100:

row00 :col00 [] 1409608777713 false 0 0
row01 :col00 [] 1409608777713 false 1 0
row02 :col00 [] 1409608777713 false 2 0
...
row98 :col00 [] 1409608777713 false 98 0
row99 :col00 [] 1409608777713 false 99 0
items returned: 100

Table options

The columns of Hive tables are mapped to columns in Accumulo when an Accumulo table is created. One of the original columns of the Hive table can be used as the row ID of the Accumulo table. This will enable queries over ranges of values for this column to be performed efficiently.

If a Hive table is to be created with columns named name, age, and height, we might choose to map these to Accumulo by storing the name column as the row ID, and the other columns in a common column family. This can be specified via the WITH SERDEPROPERTIES() function during table creation. Our use of this function can look like the following:

CREATE TABLE people(name STRING, age INT, height INT)
STORED  BY 'org.apache.hadoop.hive.accumulo.AccumuloStorageHandler'
WITH SERDEPROPERTIES('accumulo.columns.mapping' =
                     ':rowid,attributes:age,attributes:height'),

The column mapping relies on the order of fields in the Hive table specification, assigning the first Accumulo column name to the first Hive table column and so on.

By default the Accumulo table name is the same as the Hive table name. This can be overridden via the WITH TBLPROPERTIES() function:

WITH TBLPROPERTIES ("accumulo.table.name" = "hive_people")

A table created with CREATE TABLE will be considered managed by Hive, meaning it will be created and destroyed along with the Hive table. To map a Hive table to an existing Accumulo table without tying the lifecycle of the Accumulo table to the Hive table, use the EXTERNAL keyword.

If we simply want to inform Hive of a table we have already created in Accumulo, we can do something like the following:

CREATE EXTERNAL TABLE people(name STRING, age INT, height INT)
STORED  BY 'org.apache.hadoop.hive.accumulo.AccumuloStorageHandler'
WITH SERDEPROPERTIES('accumulo.columns.mapping' =
                     ':rowid,attributes:age,attributes:height'),

Serializing values

Values in Hive tables can be of various types. By default, values are serialized as strings when stored in Accumulo. Values can also be stored using a binary serialization by adding #b to the end of a field name in the mapping.

For example, if we want to store the age field using binary serialization we can specify it as attributes:age#b. To explicitly specify string serialization, #s can be used.

Additional options

Additional behavior can be controlled via the following options:

accumulo.iterator.pushdown

Use iterators to execute filter predicates. True by default.

accumulo.default.storage

Set the serialization method to be used by default when no method is specified. The default is string.

accumulo.visibility.label

A visibility label to be applied to records written to Accumulo. By default this is empty.

accumulo.authorizations

A list of authorizations, separated by commas, to be applied when scanning Accumulo tables. Blank by default.

accumulo.composite.rowid.factory

The name of a Java class that can be used to customize behavior when constructing LazyObjects from the row ID without changing the ObjectInspector.

accumulo.composite.rowid

Apply custom parsing of the row ID column into a LazyObject.

accumulo.table.name

The name of the Accumulo table to use. By default this is the same as Hive table name.

accumulo.mock.instance

Use an instance of MockAccumulo for testing instead of an actual Accumulo instance. Default is false.

Hive example

To explore how Hive and Accumulo can be used together, we’ll import some data about storm fatalities from the National Oceanic and Atmospheric Administration’s (NOAA) National Climatic Data Center website into a table in Accumulo and manipulate the data using HQL queries to answer ad-hoc questions:

$ wget http://www1.ncdc.noaa.gov/pub/data/swdi/stormevents/csvfiles/
  StormEvents_fatalities-ftp_v1.0_d2014_c20141022.csv.gz
$ gunzip StormEvents_fatalities-ftp_v1.0_d2014_c20141022.csv.gz

We can start Hive with the options that allow it to connect to a local Accumulo instance. In this example we’re using Hortonworks’ HDP 2.1 sandbox VM:

$ hive -hiveconf accumulo.instance.name=hdp 
  -hiveconf accumulo.zookeepers=localhost -hiveconf accumulo.user.name=root 
  -hiveconf accumulo.user.pass=secret

We’ll create a regular Hive table for loading the storm fatalities data from the comma-separated value (CSV) file we downloaded. Hive does not yet support loading data directly into a native table like Accumulo, so we’ll populate a Hive table and then transfer it into Accumulo:

hive> CREATE TABLE storm_fatalities(fat_yearmonth INT, fat_day INT, fat_time INT,
      fatality_id STRING, event_id STRING, fatality_type STRING,
      fatality_date DATE, fatality_age int, fatality_sex STRING,
      fatality_location string, event_yearmonth int) row format delimited fields
      terminated by ',' stored as textfile;

hive> load data local inpath
      './StormEvents_fatalities-ftp_v1.0_d2014_c20141022.csv'
      into table storm_fatalities;
Copying data from
    file:/home/hive/StormEvents_fatalities-ftp_v1.0_d2014_c20141022.csv
Copying file: file:/home/hive/StormEvents_fatalities-ftp_v1.0_d2014_c20141022.csv
Loading data to table default.storm_fatalities

Table default.storm_fatalities stats: [numFiles=1, numRows=0, totalSize=28337,
    rawDataSize=0]
OK
Time taken: 1.72 seconds

We can run a query to check our data against the Accumulo table after loading:

hive> select avg(fatality_age) from storm_fatalities;
OK
44.875
Time taken: 13.12 seconds, Fetched: 1 row(s)

Now we’ll set up the Accumulo table, similar to the Hive table, but also specifying the mapping:

hive> CREATE TABLE acc_storm_fatalities(fat_yearmonth INT, fat_day INT,
      fat_time INT, fatality_id STRING, event_id STRING, fatality_type STRING,
      fatality_date DATE, fatality_age int, fatality_sex STRING,
      fatality_location string, event_yearmonth int)
      STORED BY 'org.apache.hadoop.hive.accumulo.AccumuloStorageHandler'
      WITH SERDEPROPERTIES('accumulo.columns.mapping' =
          'time:yearmonth,time:day,time:time,:rowid,event:id,fatality:type,
           time:date,person:age,person:sex,fatality:location,event:yearmonth'),

hdfs://sandbox.hortonworks.com:8020/user/hive/warehouse/acc_storm_fatalities
OK
Time taken: 1.542 seconds

Let’s copy our storm data into the Accumulo table:

hive> INSERT OVERWRITE TABLE acc_storm_fatalities SELECT * FROM storm_fatalities;
OK
Time taken: 8.082 seconds

Looking in the Accumulo shell we can see the data in our table:

accumulo shell -u user
Password: ******

Shell - Apache Accumulo Interactive Shell
-
- version: 1.5.1.2.1.5.0-695
-
user@hdp> tables
!METADATA
acc_storm_fatalities
test
trace

user@hdp> table acc_storm_fatalities

user@hdp acc_storm_fatalities> scan
21149 event:id []    482017
21149 event:yearmonth []    201401
21149 fatality:location []    "Vehicle/Towed Trailer"
21149 fatality:type []    "D"
21149 person:sex []    "F"
21149 time:day []    6
21149 time:time []    0
21149 time:yearmonth []    201401
21186 event:id []    482572
21186 event:yearmonth []    201401
21186 fatality:location []    "Vehicle/Towed Trailer"
21186 fatality:type []    "I"
21186 person:age []    2
21186 person:sex []    "M"
21186 time:day []    6
21186 time:time []    0
21186 time:yearmonth []    201401

We can now perform queries over our Accumulo table as we would with a regular Hive table:

hive> SELECT AVG(fatality_age) FROM acc_storm_fatalities;

Total MapReduce CPU Time Spent: 3 seconds 260 msec
OK
44.875
Time taken: 30.656 seconds, Fetched: 1 row(s)

hive> SELECT count(1),fatality_location FROM acc_storm_fatalities
      GROUP BY fatality_location;
Total MapReduce CPU Time Spent: 3 seconds 160 msec
OK
1  "Boating"
1  "Business"
4  "Camping"
1  "Golfing"
2  "Heavy Equipment/Construction"
58 "In Water"
1  "Long Span Roof"
18 "Mobile/Trailer Home"
9  "Other"
87 "Outside/Open Areas"
45 "Permanent Home"
6  "Permanent Structure"
5  "Under Tree"
100  "Vehicle/Towed Trailer"
1  FATALITY_LOCATION
Time taken: 29.797 seconds, Fetched: 15 row(s)

Optimizing Hive queries

Hive works best for ad-hoc analytical queries on data stored in a columnar format. This allows Hive to only read the columns involved in a query and leave other columns unread on disk. Because Accumulo supports locality groups, we can achieve the same performance gains as other columnar storage formats (see “Column Families” and “Locality Groups”).

In our example, we might want to put the data about personal details and time into their own locality groups, leaving all other columns in the default locality group:

user@hdp acc_storm_fatalities> getgroups
user@hdp acc_storm_fatalities> setgroups person=person time=time
user@hdp acc_storm_fatalities> getgroups
time=time
person=person

We can compact to apply our changes to the files on disk:

user@hdp acc_storm_fatalities> compact
user@hdp acc_storm_fatalities>

Enabling block caching for our tables can also assist in keeping frequently accessed data blocks in memory:

user@hdp acc_storm_fatalities> config -s table.cache.block.enable=true

A query involving the field stored in the Accumulo row ID will use a Scanner configured only over the range specified. For example, the following query scans over a single row, so it would return much faster than an entire table scan:

hive> SELECT * FROM acc_storm_fatalities where event_id=500101;
OK
201404 3 0 22298 500101  "I" NULL  2 "M"
    "Permanent Home" 201404
Time taken: 0.041 seconds, Fetched: 1 row(s)

Additional notes on the Accumulo-Hive integration are available online:

Pig example

To communicate with Accumulo, Pig needs to know the location of the Accumulo JARs:

export PIG_CLASSPATH="$ACCUMULO_HOME/lib/*:$PIG_CLASSPATH"

We’ll use the example data provided for exploring Pig in Programming Pig by Alan Gates (O’Reilly):

wget -O NYSE_daily https://github.com/alanfgates/programmingpig/blob/master/
  data/NYSE_daily?raw=true

Let’s start Pig and load this file into a schema:

$ pig
grunt> daily = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,
               sdate:chararray, open:float, high:float, low:float, close:float,
               volume:int, adj_close:float);

We’ll use Pig to generate a field that we can use as a row ID in an Accumulo table; in this case we’ll use the symbol name followed by the date. This will give us a table that supports efficiently looking up all information for a particular symbol in chronological order. We’ll also generate a new field representing the closing price times the volume:

grunt> daily_by_symbol_date = foreach daily generate CONCAT(symbol, sdate), open,
       high, low, close, volume * close;

Now we can tell Pig to store this data in a table in our local Accumulo instance:

grunt> store daily_by_symbol_date
       INTO 'accumulo://daily?instance=hdp&user=root&password=secret
       &zookeepers=localhost'
       USING org.apache.pig.backend.hadoop.accumulo.AccumuloStorage('prices:open,
       prices:high,prices:low,prices:close,calculated:voltimesclose'),

Note that we didn’t mention the first element of the tuples in daily_by_symbol_date in our column specification, because that element will be written to Accumulo as the row ID:

Success!

Job Stats (time in seconds):
JobId  Alias Feature Outputs
job_local1392295495_0002 daily,daily_by_symbol_date  MAP_ONLY
  accumulo://daily?instance=hdp&user=root&password=secret&zookeepers=localhost,

Input(s):
Successfully read records from: "file:///root/pig-0.13.0/NYSE_daily"

Output(s):
Successfully stored records in: "accumulo://daily?instance=hdp&user=root
  &password=secret&zookeepers=localhost"

Job DAG:
job_local1392295495_0002

Now we should have some data in a table in Accumulo:

root@hdp daily> scan
CA1988-07-25 calculated:voltimesclose []    5.329194E7
CA1988-07-25 prices:close []    26.52
CA1988-07-25 prices:high []    26.88
CA1988-07-25 prices:low []    26.16
CA1988-07-25 prices:open []    26.16
CA1990-12-20 calculated:voltimesclose []    1.526916E7
CA1990-12-20 prices:close []    7.6
CA1990-12-20 prices:high []    7.6
CA1990-12-20 prices:low []    7.24
CA1990-12-20 prices:open []    7.24

We can continue to work with this data in Pig via LOAD statements. We’ll use the ability to scan a particular range to limit our data set to information for all dates for a single stock. We can also select just a subset of the columns available. In this example, we’ll look at only the stock CSX, and the close and voltimesclose columns:

csxinfo = LOAD 'accumulo://daily?instance=hdp&user=root&password=secret
         &zookeepers=localhost'
         USING org.apache.pig.backend.hadoop.accumulo.AccumuloStorage(
         'prices:close,calculated:voltimesclose', '-s CSX -e CSY') AS
         (symdate:chararray, close:float, voltimesclose:float);

dump csxinfo;

(CSX1988-02-18,30.12,6.0517104E7)
(CSX1988-03-02,29.25,5.87691E7)
...
(CSX2009-12-30,49.12,6.65576E7)
(CSX2009-12-31,48.49,8.255908E7)

This is much more efficient than having to read all the data and use Pig’s filter operator to limit the data.

If we want to allow Pig to use an irregular set of columns in Accumulo rows, we can use Pig maps to store whatever columns we happen to find in an Accumulo row:

csxinfo = LOAD 'accumulo://daily?instance=hdp&user=root&password=secret
         &zookeepers=localhost'
         USING org.apache.pig.backend.hadoop.accumulo.AccumuloStorage(
         '*', '-s CSX -e CSY') AS
         (symdate:chararray, allcolumns:map[]);

We can also limit the map to just those columns in a particular family:

csxinfo = LOAD 'accumulo://daily?instance=hdp&user=root&password=secret
         &zookeepers=localhost'
         USING org.apache.pig.backend.hadoop.accumulo.AccumuloStorage(
         'prices:', '-s CSX -e CSY') AS
         (symdate:chararray, prices:map[]);

As we did with Hive, we can use Accumulo’s locality groups feature to partition groups of columns into separate files on disk to make reading a particular subset of columns more efficient. Carefully generating row IDs can make accessing a subset of rows as Pig tuples dramatically more efficient.