Implementing Paging

By default, scanners return key-value pairs until the set of results is exhausted. Applications that want to enable users to page through results have several options.

For example, we can create a method that takes a start row ID, a set of columns, a page offset, and a page size:

public List<Entry<Key,Value>> getResults(String startRow, List<Text> columns,
    int offset, int pageSize)

We can choose to create brand-new Scanner objects every time this method is called, and skip over the previous page until we reach the specified offset. This has the disadvantage of having to read more and more results off of disk and transfer them to the client as the page offset increases. If users will typically only look at the first few pages this might be acceptable.

An example of using Google’s Guava library to modify a Java iterator returned from a Scanner is as follows:

import com.google.common.collect.FluentIterable;

public List<Entry<Key,Value>> getResults(String startRow, List<Text> columns,
    int offset, int pageSize) {

  // ... after the scanner has been setup
  FluentIterable<Entry<Key, Value>> fiter = FluentIterable.from(scanner);
  fiter.skip(offset);
  fiter.limit(pageSize);

  return Lists.newArrayList(fiter);
}

Another option is to cache recently created scanners and associate them with individual queries. When users request the next page in a set of results, we can simply retrieve the scanner and continue fetching the next page of key-value pairs. Scanners do not have the ability to seek backward, but if the primary method of paging through results is to start at the first and move through the pages sequentially, this method may work well. This has the disadvantage of having to keep scanners around and expire them after a certain amount of time or until the user closes the session.

Another option for paging forward is to, instead of caching Scanner objects, cache the last key-value pair seen and then create a new scanner, seeking to the next logical key that appears after the last key-value pair seen. This has the advantage of not requiring scanner resources to be kept open, but it can incur more overhead by creating a new scanner for every page requested.

When paging is implemented in the context of secondary indexes, we need to process record IDs retrieved from the index table that match the query criteria to identify the page of records requested, and then fetch only matching actual records for that page.

Index Partitioned by Term

One way to build a secondary index is by storing individual terms to be queried in the row ID. For example, we can retrieve Wikipedia pages that contain a given word by building a table storing the words found in article text in the Accumulo row ID and the article title as the column qualifier.

Table 8-2 recalls our WikipediaArticles table from “Data Modeling”, which used article titles as the row ID.

Now we create a secondary index that maps words appearing in Wikipedia pages to the page titles, shown in Table 8-3. An index organized by storing words or terms as the row ID is referred to as a term-partitioned index.

The entries for an index on a subset of Wikipedia articles are as follows:

white contents:Friendship_Games []
white contents:Olympic_Games []
white contents:Olympic_Games_ceremony []
whitfield, contents:Cotswold_Olimpick_Games []
whitsun, contents:Cotswold_Olimpick_Games []
whitsun. contents:Cotswold_Olimpick_Games []
who contents:Alternate_Olympics []
who contents:Ancient_Olympic_Games []
who contents:Arena_X-Glide []

Because in a secondary index table we’re swapping the order of the row IDs and values from the original table, an index like this is sometimes called an inverted index. However, note that we don’t store the title from the WikipediaArticles table in the value portion of the secondary index, but rather we store titles in the column qualifier. This is because a term can appear in more than one article. We don’t want article titles to be different versions of values for the terms, and we could envision wanting to scan a range of titles within a term, so we simply store the titles under the column qualifier and leave the value blank.

This technique also works for indexing the article metadata fields. It is possible to store the index entries for all fields in the same table if we want (see Table 8-4). We’ll store the concatenated column family and qualifier from the original table in the column family of the index so that a client can fetch values from a particular column, if they so choose. By not specifying the column family, a client can find the rows in which a given value appears in any field.

An example of an ingest client that writes data to the WikipediaArticles table and the WikipediaIndex table at the same time is as follows:

// write article data to articles table as before
String wikitext = page.getText();
String plaintext = model.render(converter, wikitext);
plaintext = plaintext
.replace("{{",  " ")
.replace("}}", " ");

Mutation m = new Mutation(page.getTitle()); 
m.put(WikipediaConstants.CONTENTS_FAMILY, "", plaintext);
m.put(WikipediaConstants.METADATA_FAMILY, WikipediaConstants.NAMESPACE_QUAL,
    page.getNamespace());
m.put(WikipediaConstants.METADATA_FAMILY, WikipediaConstants.TIMESTAMP_QUAL,
    page.getTimeStamp());
m.put(WikipediaConstants.METADATA_FAMILY, WikipediaConstants.ID_QUAL,
    page.getId());
m.put(WikipediaConstants.METADATA_FAMILY, WikipediaConstants.REVISION_QUAL,
    page.getRevisionId());

writer.addMutation(m);

// write index entries as well

// tokenize article contents on whitespace and punctuation and set to lowercase
HashSet<String> tokens = Sets.newHashSet(plaintext.replace(""", "")
    .toLowerCase().split("\s+"));
for (String token : tokens) {
  if (token.length() < 2) { // skip single letters
    continue;
  }

  Mutation indexMutation = new Mutation(token);
  indexMutation.put(WikipediaConstants.CONTENTS_FAMILY, page.getTitle(),
      BLANK_VALUE); 

  indexWriter.addMutation(indexMutation);
}

Create a new Mutation with the term that users can query as the row ID.

Designate this index entry as being from the article contents by specifying a column family. Store the page title in the column qualifier so that we use it to perform a subsequent lookup on the primary articles table.

We’re only indexing simple words here, by tokenizing the original text on whitespace. We talk about how to index other types of values in “Indexing Data Types”.

Now we have a table containing original articles, with the article title as the key, and another table containing index entries of words found in articles with pointers to the article titles from which they came.

Querying a Term-Partitioned Index

With this term-partitioned secondary index we can now look up article titles by the value of any metadata field, or by any word appearing in the article body. Once we have some article titles retrieved from the index table, we can retrieve the information about the articles by doing lookups against the original WikipediaArticles table using a BatchScanner:

public void querySingleTerm(String term) throws TableNotFoundException {

  Scanner scanner = conn.createScanner(WikipediaConstants.INDEX_TABLE, auths);
  // lookup term in index
  scanner.setRange(Range.exact(term));

  // store all article titles returned
  HashSet<Range> matches = new HashSet<>();
  for (Entry<Key, Value> entry : scanner) {
    matches.add(new Range(entry.getKey().getColumnQualifier().toString()));
  }

  if(matches.isEmpty()) {
    System.out.println("no results");
    return;
  }

  for (Entry<Key, Value> entry : retrieveRecords(conn, matches)) {
    System.out.println("Title:	" + entry.getKey().getRow().toString()
            + "
Revision:	" + entry.getValue().toString() + "
");
  }
}

private Iterable<Entry<Key,Value>> retrieveRecords(Connector conn,
    Collection<Range> matches) throws TableNotFoundException {
  // retrieve original articles
  BatchScanner bscanner = conn.createBatchScanner(
      WikipediaConstants.ARTICLES_TABLE, auths, 10);
  bscanner.setRanges(matches);

  // fetch only the article contents
  bscanner.fetchColumn(new Text(WikipediaConstants.METADATA_FAMILY),
          new Text(WikipediaConstants.REVISION_QUAL));

  return bscanner;
}

Note that our query code is coupled with our ingest code. If we change our ingest code, the schema of our index or original articles table will change and our query code will have to be updated in order to query these tables properly.

The query we just performed is an example of a point query, in which we find all records containing an exact term. We can also use this index to perform range queries, in which we retrieve all records matching a range of terms. See “Using Lexicoders in indexing” for an example.

Dealing with the nuances of secondary indexing in applications will be new to developers accustomed to working with relational databases, which do the work of building secondary indexes for applications. The trade-off for having to do this work is an incredible amount of flexibility in how data is indexed and retrieved.

This level of control is appropriate for Accumulo because the large data volumes Accumulo is designed to manage make it imperative for data to be organized in ways that are optimized for specific access patterns; otherwise performance will quickly degrade.

A suboptimal query on even a few gigabytes of data, such as a simple linear scan, can still be done quickly because that much data will fit comfortably in the main memory a single server and even a single desktop or notebook computer. But sub-optimal queries on hundreds of terabytes of data will be too slow for users to tolerate.

One way to view indexing is as a way to precompute views of the data that are optimal for the required access patterns. The scalability of the system and the relatively low cost of storage makes materializing these views feasible.

// returns records matching any term
public void queryMultipleTerms(String ... terms) throws TableNotFoundException {

  HashSet<String> matchingRecordIDs = new HashSet<>();

  for(String term : terms) {

    Scanner scanner = conn.createScanner(WikipediaConstants.INDEX_TABLE, auths);
    // lookup term in index
    scanner.setRange(Range.exact(term));

    for (Entry<Key, Value> entry : scanner) {
      matchingRecordIDs.add(entry.getKey().getColumnQualifier().toString());
    }
  }

  if(matchingRecordIDs.isEmpty()) {
    System.out.println("no results");
    return;
  }

  // convert to Ranges
  List<Range> ranges = Lists.newArrayList(
          Iterables.transform(matchingRecordIDs, new StringToRange()));

  for (Entry<Key, Value> entry : retrieveRecords(conn, ranges)) {
    System.out.println("Title:	" + entry.getKey().getRow().toString()
            + "
Revision:	" + entry.getValue().toString() + "
");
  }
}

private class StringToRange implements Function<String,Range> {

  @Override
  public Range apply(String f) {
    return new Range(f);
  }
}

Scanner scanner = conn.createScanner(WikipediaConstants.INDEX_TABLE, auths);
// lookup term and field in index
scanner.setRange(Range.exact(term), WikipediaConstants.CONTENTS_FAMILY); 

// store all article titles returned
HashSet<Range> matches = new HashSet<>();
for (Entry<Key, Value> entry : scanner) {
  matches.add(new Range(entry.getKey().getColumnQualifier().toString()));
}

if(matches.isEmpty()) {
  System.out.println("no results");
  return;
}

for (Entry<Key, Value> entry : retrieveRecords(conn, matches)) {
  System.out.println("Title:	" + entry.getKey().getRow().toString()
          + "
Revision:	" + entry.getValue().toString() + "
");
}

Maintaining Consistency Across Tables

Term-based secondary indexes must be maintained along with the original table so that inconsistencies do not arise. Even though Accumulo does not provide multirow transactions or cross-table transactions, this consistency can often be managed in the application.

One strategy for managing consistency between the original table and the secondary index table is to carefully order read and write operations. You can choose to wait until new rows are written to the original table first, and then write the corresponding entries to the secondary index. If for some reason a write to the original table fails, it can be retried before any index entries are written. This way clients aren’t referred by the index to a row in the original table that doesn’t exist.

Inversely, when data is deleted from the original table, the index entries should be removed first, and then the row from the original table. These strategies will prevent any clients from looking up data in the index that has not yet been written or that has been removed from the original table.

More complicated strategies may be required if an application involves concurrent updates to indexed data. One potential way to address updating secondary indexes is to look to higher-level abstractions built on top of Accumulo, such as the Fluo framework, which allows writes to be triggered to index tables from updates to a primary record table.

BatchWriterConfig conf = new BatchWriterConfig();
MultiTableBatchWriter multiTableBatchWriter =
    conn.createMultiTableBatchWriter(conf);

writer = multiTableBatchWriter.getBatchWriter(WikipediaConstants.ARTICLES_TABLE);
indexWriter =
    multiTableBatchWriter.getBatchWriter(WikipediaConstants.INDEX_TABLE);

try {
  multiTableBatchWriter.flush();
} catch (MutationsRejectedException mre) {
  // report or retry
}

try {
  multiTableBatchWriter.close();
  System.out.println("done.");
} catch (MutationsRejectedException mre) {
  // report or retry
}

Index Partitioned by Document

A basic term-partitioned index is useful for retrieving all the data containing a particular word or having a specific value for a field. If we need to find all the data containing two different words, the client code would have to issue two scans to the basic index, bringing the document IDs for both back to the client side and intersecting the two lists. This can be inefficient if one or both of the terms appears in many documents, requiring many IDs to be retrieved. One solution to this problem is to build a document-partitioned index. In such an index, sets of documents are grouped together into partitions, and each partition is assigned an ID. The index is organized first by partition ID, then by word. Table 8-6 shows an example.

The partition ID is the row portion of the key. The page contents are stored in one column family of the row, and the index is stored in another column family. To retrieve all the pages containing the words “wrestling” and “medal” in this partition, we can read over and merge the sorted lists of page titles obtained by scanning over the keys starting with partition ID_ : index : wrestling and starting with partition ID : index : medal.

This intersection can be accomplished on the server side with an appropriate Accumulo iterator. We discuss iterators in more depth in “Iterators”. An iterator that seeks to multiple starting points and intersects the results is called an intersecting iterator.

To use this method, a data set should be divided into an appropriate number of partitions so that the partitions are not too large or too small, and there are enough of them that they are spread over the desired number of servers. Ideally each partition will fill an entire tablet, so its size should be somewhere between 256 MB and tens of gigabytes. For the Wikipedia data, we’ll use 32 partitions.

Example code for building this table is as follows:

private static final int NUM_PARTITIONS = 10;
private static final Value BLANK_VALUE = new Value("".getBytes());

@Override
public void process(WikiArticle article, Siteinfo info) throws SAXException {

  String wikitext = article.getText();
  String plaintext = model.render(converter, wikitext);
  plaintext = plaintext.replace("{{", " ").replace("}}", " ");

  Mutation m = new Mutation(Integer.toString(Math.abs(
      article.getTitle().hashCode()) % NUM_PARTITIONS));
  m.put("doc" + '' + "wikiDoc", article.getTitle(), plaintext);

  // tokenize article contents on whitespace and punctuation and set to lowercase
  HashSet<String> tokens = Sets.newHashSet(plaintext.toLowerCase()
      .split("\s+"));
  for (String token : tokens) {
    m.put("ind", token + '' + "wikiDoc" + '' + article.getTitle() + '',
        BLANK_VALUE);
  }

  try {
    writer.addMutation(m);
  } catch (MutationsRejectedException e) {
    throw new SAXException(e);
  }
}

Key-value pairs in this table look as follows:

root@miniInstance> table WikipediaPartitioned
table WikipediaPartitioned
root@miniInstance WikipediaPartitioned> scan
scan
0 docx00wikiDoc:Sqoop []     Infobox software Sqoop is a ...
0 ind:x00wikiDocx00Sqoopx00 []
0 ind:2012.x00wikiDocx00Sqoopx00 []
0 ind:ax00wikiDocx00Sqoopx00 []
0 ind:accumulox00wikiDocx00Sqoopx00 []
0 ind:alsox00wikiDocx00Sqoopx00 []
0 ind:andx00wikiDocx00Sqoopx00 []
0 ind:apachex00wikiDocx00Sqoopx00 []
0 ind:applicationx00wikiDocx00Sqoopx00 []
0 ind:archivesx00wikiDocx00Sqoopx00 []

Querying a Document-Partitioned Index

When querying the data, we will use a BatchScanner along with an intersecting iterator, the IndexedDocIterator, to find relevant pages in each of the partitions. To scan all partitions, we give the BatchScanner a special range that covers the entire table.

Code to query our document-partitioned index is as follows:

BatchScanner scanner = conn.createBatchScanner(
    WikipediaConstants.DOC_PARTITIONED_TABLE, auths, 10);
scanner.setTimeout(1, TimeUnit.MINUTES);
scanner.setRanges(Collections.singleton(new Range()));

Text[] termTexts = new Text[terms.length];
for (int i = 0; i < terms.length; i++) {
  termTexts[i] = new Text(terms[i]);
}

// lookup all articles containing the terms
IteratorSetting is = new IteratorSetting(50, IndexedDocIterator.class);
IndexedDocIterator.setColfs(is, "ind", "doc");
IndexedDocIterator.setColumnFamilies(is, termTexts);
scanner.addScanIterator(is);

for (Entry<Key, Value> entry : scanner) {
  String[] parts = entry.getKey().getColumnQualifier().toString().split("");
  System.out.println(
          "doctype: " + parts[0] +
          "
docID:" + parts[1] +
          "
info: " + parts[2] +
          "

text: " + entry.getValue().toString());
}

See the WikipediaQueryMultiterm.java file for more detail.

Applications that utilize a document-partitioned index don’t necessarily need to query all partitions for every user request. For example, an application designer might choose to use partitions to implement paging and to return the first page or pages to users by scanning a subset of the partitions. Users can then request additional pages, which are populated via scans of the remaining partitions.

The mapping of documents to partitions is typically done via hashing or round-robin, but it can be done in other ways, depending on the needs of the application. For example, the value of a particular field within a document or record—such as the document type—might be chosen to determine in which partition a document or record belongs. However, care should be taken to ensure partitions are all about the same size, so that tablet servers are evenly loaded.

Term-partitioned and document-partitioned indexes are two of the more popular table designs for addressing a wide variety of access patterns with a minimum number of tables.

Indexing Data Types

Values in the original table can be just about anything. Accumulo will never interpret a value and doesn’t sort them. When building a secondary index, sort order of these items must be considered. For values to sort properly, they may need to be transformed. Here are a few examples of how the human-readable string representations of these values may not be the right way to store values in the keys of an index table.

String representations of numbers, when sorted lexicographically as Accumulo sorts them, do not end up in numerical order. These must be transformed in order to sort properly. One way to make lexicographic sorting match numeric sorting is to pad the numbers to a fixed width with zeros on their left. For example:

  0
  1
  11
  2

might be stored as:

  00
  01
  02
  11

Another example is IP addresses, which consist of four 8-bit numbers called octets, each of which ranges between 0 and 255, separated by a period. Because the string representation of an octet can be either one, two, or three characters, IP addresses do not sort well lexicographically:

192.168.1.1
192.168.1.15
192.168.1.16
192.168.1.2
192.168.1.234
192.168.1.25
192.168.1.3
192.168.1.5
192.168.1.51
192.168.1.52

To avoid this situation, the octets can simply be zero-padded to sort IP addresses properly:

192.168.001.001
192.168.001.002
192.168.001.003
192.168.001.005
192.168.001.015
192.168.001.016
192.168.001.025
192.168.001.051
192.168.001.052
192.168.001.234

Fortunately, there is a human-readable way to store dates that sorts them in the proper order using the longest time periods first and zero-padding:

YYYYMMDD

20120101
20120102
20120201
20120211
20120301

YYYYMMDDHHmmSS

Including dashes, spaces, and colons will not change this basic order:

YYYY-DD-MM HH:mm:SS

Dates could also be converted to a value such as milliseconds since midnight January 1, 1970 or some other convention and stored as numbers with appropriate padding or encoding.

In the original BigTable paper the authors describe a method for storing domain names so that subdomains that share a common domain suffix sort together:

com.google.appengine
com.google.mail
com.google.www
com.msdn
com.msdn.developers
com.yahoo
com.yahoo.mail
com.yahoo.search

It may be preferable to simply transform strings from the natural output of the toString() representation to a string that sorts values properly. If at all possible, the Lexicoder framework (described in the next section) should be used to help do this sorting, but in general knowing how to sort values is important to developing tables that allow for range queries.

Date d = dateFormat.parse(page.getTimeStamp());

byte[] dateBytes = dateLexicoder.encode(d);

Mutation dateIndexMutation = new Mutation(dateBytes);
dateIndexMutation.put(WikipediaConstants.TIMESTAMP_QUAL, page.getTitle(),
    BLANK_VALUE);
indexWriter.addMutation(dateIndexMutation);

public void queryDateRange(
          final Date start,
          final Date stop) throws TableNotFoundException {

  DateLexicoder dl = new DateLexicoder();

  Scanner scanner = conn.createScanner(WikipediaConstants.INDEX_TABLE, auths);

  // scan over the range of dates specified
  scanner.setRange(
          new Range(
            new Text(dl.encode(start)),
            new Text(dl.encode(stop))));

  // store all article titles returned
  HashSet<Range> matches = new HashSet<>();
  for (Entry<Key, Value> entry : scanner) {
    matches.add(new Range(entry.getKey().getColumnQualifier().toString()));
  }

  if(matches.isEmpty()) {
    System.out.println("no results");
    return;
  }

  for (Entry<Key, Value> entry : retrieveRecords(conn, matches)) {
    System.out.println("Title:	" + entry.getKey().getRow().toString()
            + "
Revision:	" + entry.getValue().toString() + "
");
  }
}

SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
System.out.println("querying for articles from 2015-01-01 to 2016-01-01");
query.queryDateRange(df.parse("2015-01-01"), df.parse("2016-01-01"));

querying for articles from 2015-01-01 to 2016-01-01
...
Title: Apache Hadoop
Revision:  11630810

Title: Apache Hive
Revision:  18882023

byte[] encode(V v);

V decode(byte[] b) throws ValueFormatException;

192.168.1.1
192.168.1.2
192.168.11.1
192.168.11.11
192.168.11.100
192.168.11.101
192.168.100.1
192.168.100.2
192.168.100.12

public class Inet4AddressLexicoder implements Lexicoder<Inet4Address> {

  @Override
  public byte[] encode(Inet4Address v) {
    return v.getAddress();
  }

  @Override
  public Inet4Address decode(byte[] b) throws ValueFormatException {
    try {
      return (Inet4Address) Inet4Address.getByAddress(b);
    } catch (UnknownHostException ex) {
      throw new ValueFormatException(ex.getMessage());
    }
  }
}

Connector conn = ExampleMiniCluster.getConnector();

List<String> addrs = new ArrayList<>();

addrs.add("192.168.1.1");
addrs.add("192.168.1.2");
addrs.add("192.168.11.1");
addrs.add("192.168.11.11");
addrs.add("192.168.11.100");
addrs.add("192.168.11.101");
addrs.add("192.168.100.1");
addrs.add("192.168.100.2");
addrs.add("192.168.100.12");

conn.tableOperations().create("addresses");

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

// ingest using just address strings
for(String addrString : addrs) {

  Mutation m = new Mutation(addrString);
  m.put("", "address string", addrString);

  writer.addMutation(m);
}

writer.flush();

System.out.println("sort order using strings");
Scanner scanner = conn.createScanner("addresses", Authorizations.EMPTY);
for(Map.Entry<Key, Value> e : scanner) {
  System.out.println(e.getValue());
}

sort order using strings
192.168.1.1
192.168.1.2
192.168.100.1
192.168.100.12
192.168.100.2
192.168.11.1
192.168.11.100
192.168.11.101
192.168.11.11

// delete rows
conn.tableOperations().deleteRows("addresses", null, null);


// ingest using lexicoder
Inet4AddressLexicoder lexicoder = new Inet4AddressLexicoder();

for(String addrString : addrs) {

  InetAddress addr = InetAddresses.forString(addrString);

  byte[] addrBytes = lexicoder.encode((Inet4Address)addr);

  Mutation m = new Mutation(addrBytes);
  m.put("", "address string", addrString);

  writer.addMutation(m);
}

writer.close();

// scan again
System.out.println("
sort order using lexicoder");
for(Map.Entry<Key, Value> e : scanner) {
  System.out.println(e.getValue());
}

sort order using lexicoder
192.168.1.1
192.168.1.2
192.168.11.1
192.168.11.11
192.168.11.100
192.168.11.101
192.168.100.1
192.168.100.2
192.168.100.12

wikipediaMetadata

The wikipediaMetadata table (Table 8-7) keeps track of the fields that are indexed. It is consulted in order to determine if a query requires searching fields that are not indexed. If so, the query will proceed without trying to consult the index entries in the other tables.

This table has a SummingCombiner iterator configured to add up the values of the f column family.

wikipediaIndex

The wikipediaIndex table (Table 8-8) serves as a global index, identifying which partitions contain articles that have a specified field value for a specified field name. This is so that partitions that don’t contain any information about a particular search term can be omitted from the set of partitions to query in the second step.

This table has an additional iterator configured, the GlobalIndexUidCombiner. This iterator maintains a list of article IDs that are associated with a search term and a count of how many times this search term has been written to this table. If the list of IDs grows over 20 by default, then it stops keeping track of individual UIDs and only keeps the count.

This table is queried to obtain information on the number of articles in which a search term appears and optionally, if the number of articles is low enough, the actual list of article IDs in which a search term appears. In these cases, this saves us an additional lookup against the wikipedia table.

Once the information about all the search terms in a query has been obtained from this table, the query logic determines whether to do additional scans against the wikipedia table, and what type of scans to do—whether an optimized scan including the index within each partition searched or a full table scan.

wikipedia

The wikipedia table (Table 8-9) contains the full text of each article and a set of index entries. The set of documents within a partition appears first, under the d column family. Then there are a set of index entries consisting of a column family beginning with the prefix fi and containing the field name in which a term appears, a column qualifier containing the word found in the field, the language ID, and the article ID.

This is organized to allow a server-side iterator to scan over the index entries and determine which articles satisfy all of the query criteria specified. Once a set of articles is obtained the iterator can then return either the content for the set of matching documents or simply the article IDs.

This table doesn’t have any iterators configured after ingest, but when the query code determines that an optimized query plan can be executed, the OptimizedQueryIterator class or EvaluatingIterator class can be applied to BatchScanner objects and configured for the duration of a particular query.

wikipediaReverseIndex

The wikipediaReverseIndex table (Table 8-10) is the reverse of the wikipediaIndex table. It is used to perform index lookups using leading wildcards—instead of the wikipediaIndex table, which supports exact term matches and those with trailing wildcards.

Ingesting WikiSearch Data

We’ll work through installing and using the wikisearch project and examine the tables created:

 [accumulo@host ~]$ git clone 
     https://git-wip-us.apache.org/repos/asf/accumulo-wikisearch.git

Next we’ll copy the example configuration file and edit it to work with our Accumulo instance:

 [accumulo@host ~]$ cd accumulo-wikisearch
 [accumulo@host accumulo-wikisearch]$ mvn
 [accumulo@host accumulo-wikisearch]$ cp ingest/conf/wikipedia.xml.example 
     conf/wikipedia.xml
 [accumulo@host accumulo-wikisearch]$ vi ingest/conf/wikipedia.xml

The configuration file should be filled in with the information about our Accumulo cluster:

<configuration>
  <property>
    <name>wikipedia.accumulo.zookeepers</name>
    <value>your-zookeeper:2181</value>
  </property>
  <property>
    <name>wikipedia.accumulo.instance_name</name>
    <value>your-instance</value>
  </property>
  <property>
    <name>wikipedia.accumulo.user</name>
    <value>your-username</value>
  </property>
  <property>
    <name>wikipedia.accumulo.password</name>
    <value>your-password</value>
  </property>
  <property>
    <name>wikipedia.accumulo.table</name>
    <value>wikipedia</value>
  </property>
  <property>
    <name>wikipedia.ingest.partitions</name>
    <value>5</value>
  </property>
</configuration>

With the configuration file set up the way we want it, we need to install the project’s iterators to a location where the tablet servers can access and load them. In this case we’ll use the $ACCUMULO_HOME/lib/ext/ directory on the local filesystem of each of the tablet servers:

[accumulo@host accumulo-wikisearch]$ scp ingest/lib/wikisearch-ingest-1.5.0.jar 
    accumulo@tserver1:/opt/accumulo/accumulo-1.5.0/lib/ext/
[accumulo@host accumulo-wikisearch]$ scp ingest/lib/protobuf-java-2.3.0.jar 
    accumulo@tserver1:/opt/accumulo/accumulo-1.5.0/lib/ext/

Now we’ll place a file containing some Wikipedia articles into HDFS so they can be loaded into Accumulo via a MapReduce job. See “Wikipedia Data” for details on obtaining Wikipedia files:

[accumulo@host ~]$ mv enwiki-latest-pages-articles1.xml-p000000010p000010000.bz2
    wiki.xml.bz2
[accumulo@host ~]$ hdfs dfs -mkdir /input
[accumulo@host ~]$ hdfs dfs -mkdir /input/wiki
[accumulo@host ~]$ hdfs dfs  -put wiki.xml.bz2 /input/wiki/

Now we’re ready to run the script that loads this data into tables in Accumulo:

[accumulo@host accumulo-wikisearch]$ cd ingest/bin
[accumulo@host bin]$ ./ingest.sh /input/wiki/

INFO zookeeper.ClientCnxn: Session establishment complete on server
    zookeeper:2181
Input files in /input/wiki: 1
Languages:1

INFO input.FileInputFormat: Total input paths to process : 1
INFO mapred.JobClient: Running job: job_201410202349_0007
INFO mapred.JobClient:  map 0% reduce 0%
INFO mapred.JobClient:  map 100% reduce 0%
INFO mapred.JobClient: Job complete: job_201410202349_0007

When the job is complete we can examine the tables. The import code applies security tokens for the language of an article to the key-value pairs imported, so we need to grant these tokens to our Accumulo user:

[accumulo@host ~]$ accumulo shell -u accumulo
password:

accumulo@host> setauths -u <user> -s all,enwiki,eswiki,frwiki,fawiki
accumulo@host> tables
!METADATA
trace
wikipedia
wikipediaIndex
wikipediaMetadata
wikipediaReverseIndex

We’ll set up an application to query these tables in the next section.

Querying the WikiSearch Data

This example project ships with a web application that we can use to query the wiki-search tables we created in the preceding section.

First, we’ll configure the app for our Accumulo instance:

[accumulo@host query]$ cp src/main/resources/META-INF/ejb-jar.xml.example 
    src/main/resources/META-INF/ejb-jar.xml
[accumulo@host query]$ vi src/main/resources/META-INF/ejb-jar.xml

<enterprise-beans>
    <session>
      <ejb-name>Query</ejb-name>
      <env-entry>
        <env-entry-name>instanceName</env-entry-name>
        <env-entry-type>java.lang.String</env-entry-type>
        <env-entry-value>your-instance</env-entry-value>
      </env-entry>
      <env-entry>
        <env-entry-name>zooKeepers</env-entry-name>
        <env-entry-type>java.lang.String</env-entry-type>
        <env-entry-value>your-zookeepers</env-entry-value>
      </env-entry>
      <env-entry>
        <env-entry-name>username</env-entry-name>
        <env-entry-type>java.lang.String</env-entry-type>
        <env-entry-value>your-username</env-entry-value>
      </env-entry>
      <env-entry>
        <env-entry-name>password</env-entry-name>
        <env-entry-type>java.lang.String</env-entry-type>
        <env-entry-value>your-password</env-entry-value>
      </env-entry>
      <env-entry>
        <env-entry-name>tableName</env-entry-name>
        <env-entry-type>java.lang.String</env-entry-type>
        <env-entry-value>wikipedia</env-entry-value>
      </env-entry>
      <env-entry>
        <env-entry-name>partitions</env-entry-name>
        <env-entry-type>java.lang.Integer</env-entry-type>
        <env-entry-value>5</env-entry-value>
      </env-entry>
      <env-entry>
        <env-entry-name>threads</env-entry-name>
        <env-entry-type>java.lang.Integer</env-entry-type>
        <env-entry-value>8</env-entry-value>
      </env-entry>
    </session>
  </enterprise-beans>

Next, we’ll build the query project:

[accumulo@host accumulo-wikisearch]$ mvn install
[accumulo@host accumulo-wikisearch]$ cd query
[accumulo@host query]$ mvn package assembly:single

Now we’ll install it in a JBoss AS 6.1 server. In our case JBoss lives in /opt:

 [accumulo@host query]$ cd /opt/jboss/server/default
 [accumulo@host default]$ tar -xzf ~/accumulo-wikisearch/query/target/
     wikisearch-query-1.5.0-dist.tar.gz

Copy over the WAR file to the deploy directory:

 [accumulo@host default]$ cp ~/accumulo-wikisearch/query-war/target/
     wikisearch-query-war-1.5.0.war deploy/

Now we can start JBoss:

 [accumulo@host deploy]$ /opt/jboss/bin/run.sh -b 0.0.0.0 &

Finally, we’ll copy some JAR files from JBoss’s directories into the lib/ext/ directories of our tablet servers:

 [accumulo@host lib]$ sudo cp kryo-1.04.jar 
     /opt/accumulo/accumulo-1.5.0/lib/ext/
 [accumulo@host lib]$ sudo cp minlog-1.2.jar 
     /opt/accumulo/accumulo-1.5.0/lib/ext/
 [accumulo@host lib]$ sudo cp commons-jexl-2.0.1.jar 
     /opt/accumulo/accumulo-1.5.0/lib/ext/
 [accumulo@host lib]$ cd ..
 [accumulo@host default]$ sudo cp deploy/wikisearch-query-1.5.0.jar 
     /opt/accumulo/accumulo-1.5.0/lib/ext/

We can bring up the user interface for this application by going to http://<hostname>:8080/accumulo-wikisearch/ui/ui.jsp in a web browser (Figure 8-2).

Figure 8-2. The Wikisearch UI

This example uses the Apache Commons JEXL library to create a query language. The supported JEXL operators include:

• ==

• !=

• <

• <=

• >

• >=

• =~

• !~

• and

• or

We’ll do a search for documents that contain both the words old and man:

TEXT == 'old' and TEXT == 'man'

This returns the results in Figure 8-3.

Figure 8-3. Search results

The logs show that there were 986 matching entries for this query:

HTML query: TEXT == 'old' and TEXT == 'man'
Connecting to [instanceName = koverse, zookeepers = koversevm:2181,
    username = root].
986 matching entries found in optimized query.
AbstractQueryLogic: TEXT == 'old' and TEXT == 'man' 2.63
  1) parse query 0.00
  2) query metadata 0.01
  3) full scan query 0.00
  3) optimized query 2.62
  1) process results 0.14
      1) query global index 0.02
1976233182 Query completed.

We’ll try adding another search term, sea:

TEXT == 'old' and TEXT == 'man' and TEXT == 'sea'

This returns the results in Figure 8-4.

Figure 8-4. Refined search results

This cuts down the matching entries to only 339:

HTML query: TEXT == 'old' and TEXT == 'man' and TEXT == 'sea'
Connecting to [instanceName = koverse, zookeepers = koversevm:2181,
    username = root].
339 matching entries found in optimized query.

Lexicoders

The first place to look for help in designing row IDs that sort properly is Lexicoders. Lexicoders are a set of classes designed to help convert a variety of object types into byte arrays that preserve the native sort order of the objects. We introduce Lexicoders in “Using Lexicoders in indexing” for helping sort various types of objects.

Composite Row IDs

When you construct row IDs that consist of multiple elements, it is necessary to use delimiters in order to have rows sorted hierarchically, such that all of the rows that begin with one element are sorted before the first row of the next first-place element appears.

For example, if we want to sort data by first name then secondarily by last name, we need a delimiter to ensure that if we read only first names we see them all in order regardless of what last names might follow each first name. Without a delimiter we can run into the following situation:

bobanderson
bobbyanderson
bobjones

In this case “bob anderson” should be followed by “bob jones,” but because we are missing a delimiter, “bobby anderson” appears between them. Using a delimiter we get the desired sort order:

bob_anderson
bob_jones
bobby_anderson

Because the delimiter, in this case an underscore, sorts before the third b in “bobby,” all first names with four or more letters are sorted after all the appearances of “bob.”

An effective delimiter should be a character that sorts before any characters that are likely to appear in the elements of a row ID. The null character 0 can be used if necessary.

Composite row IDs may be human-readable enough as-is. If not, custom Formatters can be written to make viewing them easier. See “Human-Readable Versus Binary Values and Formatters”.

The ListLexicoder and PairLexicoder can help in designing composite row IDs.

Key Size

As mentioned in “Constraints”, Accumulo 1.6 has a constraint added to new tables that the complete key be smaller than 1 MB. Keep in mind that the complete key includes the row ID, column family, column qualifier, visibility, and timestamp. Keeping the key under 1 MB is a best practice for all versions of Accumulo.

Avoiding Hotspots

In many table designs, hotspots can arise as the result of uneven distribution of row IDs, which often arises from skew in the source data. A common example is that of time-ordered row IDs, which we addressed by introducing a fixed number of bucket IDs as prefixes to spread newly written data across multiple parts of the table. We discuss avoiding hotspots in that we discuss in “Time-Ordered Data”.

Other examples include frequently appearing items in data sets, such as very frequent words appearing in textual data, highly populated areas in geospatial data, and temporal spikes in time series data. These can all cause an inordinate amount of data to be sent to one server, undermining the effectiveness of the distributed system.

Hotspots can involve simply one server being many times busier than all the others. They also can involve contention over individual rows and the creation of very large rows, due to Accumulo’s control over concurrent access to each individual row. In either case, the general approach is to alter the row IDs to either spread them over a larger portion of the table and therefore over a larger number of servers, or to simply break up highly contested rows into multiple rows to eliminate contention and overly large rows.

In the case of many writes ending up going to one server, introducing some sort of prefix in front of the row ID can cause the writes to be sent to as many servers as there are unique prefixes. An example of this is the fixed buckets that we discuss in “Time-Ordered Data” and also in document-partitioned indexes in “Index Partitioned by Document”.

An example of breaking up contentious rows is to append a suffix to the row ID. Multiple writes to the same data then end up going to different row IDs while keeping the rows next to each other, preserving scan order. One example is in indexing the word the, which appears more than any other word in English documents. Instead of simply indexing the word the, you can attach a random suffix to the word the like this:

the_023012
the_034231
the_323133
the_812500

This way multiple writers can still index the word the and avoid contention because they are technically writing to different rows. In addition, the tablet containing the word the can now be split into multiple tablets hosted on multiple servers, which would not be possible if all instances of the word the were indexed into the same row.

Scanners only need to be modified to begin at the first random suffix and end at the last:

'the_' to 'the`'

Other strategies for avoiding hotspots in indexed data involve indexing pairs of words when one word is very frequent. Instead of indexing the we would index the_car, the_house, etc. This has the advantage of making it easier to find records containing two words when one word is very frequent, while preserving the ability to retrieve all records containing just the frequent word.

Sometimes, very frequent items are not of interest to an application and can simply be omitted from the index. Apache Lucene and other indexing libraries often employ stop lists, which contain very frequent words that can be skipped when individual words in a document are indexed.

Some users have used Accumulo combiners to keep track of how many times a term appears in an index using a separate column family and cease to store additional terms after seeing a given number of them, as in the wikisearch example in “Full-Text Search”. This strategy is useful because it doesn’t require knowing the frequent terms beforehand, as a stop list does. However, by itself it doesn’t prevent clients from continuing to write frequent terms that will be ignored. An index like this could be scanned periodically (perhaps using a MapReduce job) to retrieve only the highly frequent terms for the purpose of creating a stop list that clients can use.

Designing Row IDs for Consistent Updates

Accumulo is designed to split tables into tablets on row boundaries. Tablet servers will not split a row into two tablets, so each row is fully contained within one tablet. The Accumulo master ensures that exactly one tablet server will be responsible for each tablet, and therefore each row. As a result, applications can make multiple changes to the data in one row simultaneously or, in database parlance, atomically, meaning that the server will never apply a portion of the changes. If something goes wrong while some changes are applied, the mutation will simply fail, the row will revert to the last consistent state, and the client process can try it again.

Applications that need to make updates to several data elements simultaneously can try to use the row construct to gather the data that needs to be changed simultaneously together under one row ID. An example is perhaps changing all of the elements of a customer address simultaneously so that the address is always valid and not some combination of an old and new address.

Sometimes grouping data that needs to be changed under a common row ID is not possible. An example is an application that needs to transfer amounts of money between accounts. This involves subtracting an amount from one account and adding it to the other account. Either both or neither of these actions should succeed. If only one succeeds, money is either created or destroyed. The pair of accounts that needs to be modified is not known beforehand and is impractical for use as a row ID.

It is possible to achieve this capability in a system based on BigTable as evidenced by Google’s Percolator paper, which describes an application layer implemented over BigTable that provides distributed atomic updates, or distributed transactions. The Fluo project is developing a framework for distributed transactions on top of Accumulo.

Also see “Transactions”.

Storing Files and Large Values

Accumulo is designed to store structured and semistructured data. It is not optimized to serve very large values, such as those that can arise from storing entire files in Accumulo. The practical limit for a value size depends on available memory, because Accumulo loads several values into memory simultaneously when servicing client requests.

When storing larger values than what comfortably fits in memory, users typically do one of two things: store the files in HDFS or some other scalable filesystem or blob store such as Amazon’s S3, or break up files into smaller chunks.

When storing files in an external filesystem or blob store, Accumulo only needs to store a pointer, such as a URL, to where the actual file can be retrieved from the external store. This has the advantage of allowing users to search and find files using Accumulo. It also inherits all the benefits of security and indexing while not having to store that actual data in Accumulo, which frees up resources for just doing lookups.

If users are more interested in retrieving specific parts of files, breaking up files into chunks and storing them in Accumulo may work better, because Accumulo can then provide the chunk of the user-requested file in one request rather than looking up the file pointer in Accumulo and fetching the file it from an external system. Files broken up into chunks can still cause problems when many chunks are retrieved simultaneously, because they can overwhelm available memory.

The Accumulo documentation includes an example of storing files as well as some discussion.

Users have contributed some example techniques for doing this.

All of this logic is managed in an application or service layer implemented above Accumulo.

Human-Readable Versus Binary Values and Formatters

In some cases it is convenient to store values in a format that is readable by humans. For example, debugging becomes easier, and viewing data in the Accumulo shell is possible.

In some cases, values are stored in human-readable form, such as UTF8 strings, and are converted to binary on the fly for operations, then converted back to human-readable values before they’re written back to the table. One example is that of storing numbers as strings in a table configured to sum the numerical values in those strings. In this case, the iterator that performs the summation is responsible for converting strings into Long or Double objects before summing them together, and then converting them back into String objects before outputting them to be sent either to the user or to the disk for storage. The provided SummingCombiner can be configured to do this for strings, or to simply treat values as Long objects.

Many applications can be made more efficient by using binary values. In this case, however, values are no longer easily read in the shell. To make debugging and viewing binary values easier, users can create a custom Formatter by implementing org.apache.accumulo.core.util.format.Formatter. This will allow the shell to display otherwise unreadable keys and values using some human-readable representation:

 package org.apache.accumulo.examples;

 /**
  * this is an example formatter that only shows a deserialized value
  * and not the key
  */

 public class ExampleFormatter implements Formatter {

  private Iterator<Entry<Key,Value>> iter;

  @Override
  public void initialize(Iterable<Map.Entry<Key, Value>> scanner,
            boolean includeTimestamps) {
    iter = scanner.iterator();
  }

  @Override
  public boolean hasNext() {
    return iter.hasNext();
  }

  @Override
  public String next() {
    Entry<Key,Value> n = iter.next();
    byte[] bytes = n.getValue().getBytes();
    // deserialize
    String s = myDeserializationFunction(bytes);
    return s;
  }

  @Override
  public void remove() {
  }

  private String myDeserializationFunction(byte[] bytes) {
    ...
  }
 }

Formatters can be configured on a per-table basis by setting the table.formatter option. Customer formatters only need to be included on the CLASSPATH when you run the shell.

The shell also makes it easy to configure formatters via the formatter command.

To add a formatter:

user@accumulo> table myTable
user@accumulo myTable> scan

user@accumulo> formatter -f org.apache.accumulo.examples.ExampleFormatter 
    -t myTable
user@accumulo> scan

To remove a formatter:

user@accumulo> formatter -r -t myTable