Apache Accumulo® User Manual Version 1.9

The TabletServer manages some subset of all the tablets (partitions of tables). This includes receiving writes from clients, persisting writes to a write-ahead log, sorting new key-value pairs in memory, periodically flushing sorted key-value pairs to new files in HDFS, and responding to reads from clients, forming a merge-sorted view of all keys and values from all the files it has created and the sorted in-memory store.

TabletServers also perform recovery of a tablet that was previously on a server that failed, reapplying any writes found in the write-ahead log to the tablet.

Accumulo processes will share files stored in HDFS. Periodically, the Garbage Collector will identify files that are no longer needed by any process, and delete them. Multiple garbage collectors can be run to provide hot-standby support. They will perform leader election among themselves to choose a single active instance.

The Accumulo Master is responsible for detecting and responding to TabletServer failure. It tries to balance the load across TabletServer by assigning tablets carefully and instructing TabletServers to unload tablets when necessary. The Master ensures all tablets are assigned to one TabletServer each, and handles table creation, alteration, and deletion requests from clients. The Master also coordinates startup, graceful shutdown and recovery of changes in write-ahead logs when Tablet servers fail.

Multiple masters may be run. The masters will choose among themselves a single master, and the others will become backups if the master should fail.

The Accumulo Tracer process supports the distributed timing API provided by Accumulo. One to many of these processes can be run on a cluster which will write the timing information to a given Accumulo table for future reference. Seeing the section on Tracing for more information on this support.

The Accumulo Monitor is a web application that provides a wealth of information about the state of an instance. The Monitor shows graphs and tables which contain information about read/write rates, cache hit/miss rates, and Accumulo table information such as scan rate and active/queued compactions. Additionally, the Monitor should always be the first point of entry when attempting to debug an Accumulo problem as it will show high-level problems in addition to aggregated errors from all nodes in the cluster. See the section on Monitoring for more information.

Multiple Monitors can be run to provide hot-standby support in the face of failure. Due to the forwarding of logs from remote hosts to the Monitor, only one Monitor process should be active at one time. Leader election will be performed internally to choose the active Monitor.

Accumulo includes a client library that is linked to every application. The client library contains logic for finding servers managing a particular tablet, and communicating with TabletServers to write and retrieve key-value pairs.

The BatchWriter is highly optimized to send Mutations to multiple TabletServers and automatically batches Mutations destined for the same TabletServer to amortize network overhead. Care must be taken to avoid changing the contents of any Object passed to the BatchWriter since it keeps objects in memory while batching.

Mutations are added to a BatchWriter thus:

An example of using the batch writer can be found at accumulo/docs/examples/README.batch.

The ConditionalWriter enables efficient, atomic read-modify-write operations on rows. The ConditionalWriter writes special Mutations which have a list of per column conditions that must all be met before the mutation is applied. The conditions are checked in the tablet server while a row lock is held (Mutations written by the BatchWriter will not obtain a row lock). The conditions that can be checked for a column are equality and absence. For example a conditional mutation can require that column A is absent inorder to be applied. Iterators can be applied when checking conditions. Using iterators, many other operations besides equality and absence can be checked. For example, using an iterator that converts values less than 5 to 0 and everything else to 1, its possible to only apply a mutation when a column is less than 5.

In the case when a tablet server dies after a client sent a conditional mutation, its not known if the mutation was applied or not. When this happens the ConditionalWriter reports a status of UNKNOWN for the ConditionalMutation. In many cases this situation can be dealt with by simply reading the row again and possibly sending another conditional mutation. If this is not sufficient, then a higher level of abstraction can be built by storing transactional information within a row.

An example of using the batch writer can be found at accumulo/docs/examples/README.reservations.

By default, Accumulo writes out any updates to the Write-Ahead Log (WAL). Every change goes into a file in HDFS and is sync’d to disk for maximum durability. In the event of a failure, writes held in memory are replayed from the WAL. Like all files in HDFS, this file is also replicated. Sending updates to the replicas, and waiting for a permanent sync to disk can significantly write speeds.

Accumulo allows users to use less tolerant forms of durability when writing. These levels are:

The user can set the default durability of a table in the shell. When writing, the user can configure the BatchWriter or ConditionalWriter to use a different level of durability for the session. This will override the default durability setting.

To retrieve data, Clients use a Scanner, which acts like an Iterator over keys and values. Scanners can be configured to start and stop at particular keys, and to return a subset of the columns available.

Accumulo supports the ability to present an isolated view of rows when scanning. There are three possible ways that a row could change in Accumulo :

Isolation guarantees that either all or none of the changes made by these operations on a row are seen. Use the IsolatedScanner to obtain an isolated view of an Accumulo table. When using the regular scanner it is possible to see a non isolated view of a row. For example if a mutation modifies three columns, it is possible that you will only see two of those modifications. With the isolated scanner either all three of the changes are seen or none.

The IsolatedScanner buffers rows on the client side so a large row will not crash a tablet server. By default rows are buffered in memory, but the user can easily supply their own buffer if they wish to buffer to disk when rows are large.

For an example, look at the following

For some types of access, it is more efficient to retrieve several ranges simultaneously. This arises when accessing a set of rows that are not consecutive whose IDs have been retrieved from a secondary index, for example.

The BatchScanner is configured similarly to the Scanner; it can be configured to retrieve a subset of the columns available, but rather than passing a single Range, BatchScanners accept a set of Ranges. It is important to note that the keys returned by a BatchScanner are not in sorted order since the keys streamed are from multiple TabletServers in parallel.

An example of the BatchScanner can be found at accumulo/docs/examples/README.batch.

The proxy server can live on any node in which the basic client API would work. That means it must be able to communicate with the Master, ZooKeepers, NameNode, and the DataNodes. A proxy client only needs the ability to communicate with the proxy server.

The configuration options for the proxy server live inside of a properties file. At the very least, you need to supply the following properties:

You can find a sample configuration file in your distribution:

This sample configuration file further demonstrates an ability to back the proxy server by MockAccumulo or the MiniAccumuloCluster.

After the properties file holding the configuration is created, the proxy server can be started using the following command in the Accumulo distribution (assuming your properties file is named config.properties):

Aside from installing the Thrift compiler, you will also need the language-specific library for Thrift installed to generate client code in that language. Typically, your operating system’s package manager will be able to automatically install these for you in an expected location such as /usr/lib/python/site-packages/thrift.

You can find the thrift file for generating the client:

After a client is generated, the port specified in the configuration properties above will be used to connect to the server.

The following examples have been written in Java and the method signatures may be slightly different depending on the language specified when generating client with the Thrift compiler. After initiating a connection to the Proxy (see Apache Thrift’s documentation for examples of connecting to a Thrift service), the methods on the proxy client will be available. The first thing to do is log in:

Once logged in, the token returned will be used for most subsequent calls to the client. Let’s create a table, add some data, scan the table, and delete it.

First, create a table.

Next, add some data:

Scan for the data and batch the return of the results on the server:

The assignment of Column Families to Locality Groups can be changed at any time. The physical movement of column families into their new locality groups takes place via the periodic Major Compaction process that takes place continuously in the background. Major Compaction can also be scheduled to take place immediately through the shell:

Iterators can be configured on a table at scan, minor compaction and/or major compaction scopes. If the Iterator implements the OptionDescriber interface, the setiter command can be used which will interactively prompt the user to provide values for the given necessary options.

The config command can always be used to manually configure iterators which is useful in cases where the Iterator does not implement the OptionDescriber interface.

Typically, a table will have multiple iterators. Accumulo configures a set of system level iterators for each table. These iterators provide core functionality like visibility label filtering and may not be removed by users. User level iterators are applied in the order of their priority. Priority is a user configured integer; iterators with lower numbers go first, passing the results of their iteration on to the other iterators up the stack.

Some iterators take additional parameters from client code, as in the following example:

Tables support separate Iterator settings to be applied at scan time, upon minor compaction and upon major compaction. For most uses, tables will have identical iterator settings for all three to avoid inconsistent results.

Accumulo provides the capability to manage versioned data through the use of timestamps within the Key. If a timestamp is not specified in the key created by the client then the system will set the timestamp to the current time. Two keys with identical rowIDs and columns but different timestamps are considered two versions of the same key. If two inserts are made into Accumulo with the same rowID, column, and timestamp, then the behavior is non-deterministic.

Timestamps are sorted in descending order, so the most recent data comes first. Accumulo can be configured to return the top k versions, or versions later than a given date. The default is to return the one most recent version.

The version policy can be changed by changing the VersioningIterator options for a table as follows:

When a table is created, by default its configured to use the VersioningIterator and keep one version. A table can be created without the VersioningIterator with the -ndi option in the shell. Also the Java API has the following method

When scanning over a set of key-value pairs it is possible to apply an arbitrary filtering policy through the use of a Filter. Filters are types of iterators that return only key-value pairs that satisfy the filter logic. Accumulo has a few built-in filters that can be configured on any table: AgeOff, ColumnAgeOff, Timestamp, NoVis, and RegEx. More can be added by writing a Java class that extends the org.apache.accumulo.core.iterators.Filter class.

The AgeOff filter can be configured to remove data older than a certain date or a fixed amount of time from the present. The following example sets a table to delete everything inserted over 30 seconds ago:

To see the iterator settings for a table, use:

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 a summing combiner were configured on a table and the following mutations were inserted:

The table would reflect only one aggregate value:

Combiners can be enabled for a table using the setiter command in the shell. Below is an example.

Accumulo includes some useful Combiners out of the box. To find these look in the org.apache.accumulo.core.iterators.user package.

Additional Combiners can be added by creating a Java class that extends org.apache.accumulo.core.iterators.Combiner and adding a jar containing that class to Accumulo’s lib/ext directory.

An example of a Combiner can be found under

The following example demonstrates Accumulo’s mechanism for exporting and importing tables.

The shell session below illustrates creating a table, inserting data, and exporting the table.

After executing the export command, a few files are created in the hdfs dir. One of the files is a list of files to distcp as shown below.

Before the table can be imported, it must be copied using distcp. After the distcp completea, the cloned table may be deleted.

The Accumulo shell session below shows importing the table and inspecting it. The data, splits, config, and logical time information for the table were preserved.

The init method is called by the TabletServer after it constructs an instance of the Iterator. This method should clear/reset any internal state in the Iterator and prepare it to process data. The first argument, the source, is the Iterator "below" this Iterator (where the client is at "top" and the Iterator for files in HDFS are at the "bottom"). The "source" Iterator provides the Key-Value pairs which this Iterator will operate upon.

The second argument, a Map of options, is made up of options provided by the user, options set in the table’s configuration, and/or options set in the containing namespace’s configuration. These options allow for Iterators to dynamically configure themselves on the fly. If no options are used in the current context (a Scan or Compaction), the Map will be empty. An example of a configuration item for an Iterator could be a pattern used to filter Key-Value pairs in a regular expression Iterator.

The third argument, the IteratorEnvironment, is a special object which provides information to this Iterator about the context in which it was invoked. Commonly, this information is not necessary to inspect. For example, if an Iterator knows that it is running in the context of a full-major compaction (reading all of the data) as opposed to a user scan (which may strongly limit the number of columns), the Iterator might make different algorithmic decisions in an attempt to optimize itself.

The seek method is likely the most confusing method on the Iterator interface. The purpose of this method is to advance the stream of Key-Value pairs to a certain point in the iteration (the Accumulo table). It is common that before the implementation of this method returns some additional processing is performed which may further advance the current position past the startKey of the Range. This, however, is dependent on the functionality the iterator provides. For example, a filtering iterator would consume a number Key-Value pairs which do not meets its criteria before seek returns. The important condition for seek to meet is that this Iterator should be ready to return the first Key-Value pair, or none if no such pair is available, when the method returns. The Key-Value pair would be returned by getTopKey and getTopValue, respectively, and hasTop should return a boolean denoting whether or not there is a Key-Value pair to return.

The arguments passed to seek are as follows:

The TabletServer first provides a Range, an object which defines some collection of Accumulo Key`s, which defines the Key-Value pairs that this Iterator should return. Each `Range has a startKey and endKey with an inclusive flag for both. While this Range is often similar to the Range(s) set by the client on a Scanner or BatchScanner, it is not guaranteed to be a Range that the client set. Accumulo will split up larger ranges and group them together based on Tablet boundaries per TabletServer. Iterators should not attempt to implement any custom logic based on the Range(s) provided to seek and Iterators should not return any Keys that fall outside of the provided Range.

The second argument, a Collection<ByteSequence>, is the set of column families which should be retained or excluded by this Iterator. The third argument, a boolean, defines whether the collection of column families should be treated as an inclusion collection (true) or an exclusion collection (false).

It is likely that all implementations of seek will first make a call to the seek method on the "source" Iterator that was provided in the init method. The collection of column families and the boolean include argument should be passed down as well as the Range. Somewhat commonly, the Iterator will also implement some sort of additional logic to find or compute the first Key-Value pair in the provided Range. For example, a regular expression Iterator would consume all records which do not match the given pattern before returning from seek.

It is important to retain the original Range passed to this method to know when this Iterator should stop reading more Key-Value pairs. Ignoring this typically does not affect scans from a Scanner, but it will result in duplicate keys emitting from a BatchScan if the scanned table has more than one tablet. Best practice is to never emit entries outside the seek range.

The next method is analogous to the next method on a Java Iterator: this method should advance the Iterator to the next Key-Value pair. For implementations that perform some filtering or complex logic, this may result in more than one Key-Value pair being inspected. This method alters some internal state that is exposed via the hasTop, getTopKey, and getTopValue methods.

The result of this method is commonly caching a Key-Value pair which getTopKey and getTopValue can later return. While there is another Key-Value pair to return, hasTop should return true. If there are no more Key-Value pairs to return from this Iterator since the last call to seek, hasTop should return false.

The hasTop method is similar to the hasNext method on a Java Iterator in that it informs the caller if there is a Key-Value pair to be returned. If there is no pair to return, this method should return false. Like a Java Iterator, multiple calls to hasTop (without calling next) should not alter the internal state of the Iterator.

These methods simply return the current Key-Value pair for this iterator. If hasTop returns true, both of these methods should return non-null objects. If hasTop returns false, it is undefined what these methods should return. Like hasTop, multiple calls to these methods should not alter the state of the Iterator.

Users should take caution when either

In both cases, copying the Key/Value’s data into a new object ensures iterator correctness. If neither case applies, it is safe to not copy the Key/Value. The general guideline is to be aware of who else may use Key/Value objects returned from getTopKey/getTopValue.

The deepCopy method is similar to the clone method from the Java Cloneable interface. Implementations of this method should return a new object of the same type as the Accumulo Iterator instance it was called on. Any internal state from the instance deepCopy was called on should be carried over to the returned copy. The returned copy should be ready to have seek called on it. The SortedKeyValueIterator interface guarantees that init will be called on an iterator before deepCopy and that init will not be called on the iterator returned by deepCopy.

Typically, implementations of deepCopy call a copy-constructor which will initialize internal data structures. As with seek, it is common for the IteratorEnvironment argument to be ignored as most Iterator implementations can be written without the explicit information the environment provides.

In the analogy of a series of Iterators representing a tree, deepCopy can be thought of as early programming assignments which implement their own tree data structures. deepCopy calls copy on its sources (the children), copies itself, attaches the copies of the children, and then returns itself.

The Filter abstract Iterator provides a very simple implementation which allows implementations to define whether or not a Key-Value pair should be returned via an accept(Key, Value) method.

Filters are extremely simple to implement; however, when the implementation is filtering a large percentage of Key-Value pairs with respect to the total number of pairs examined, it can be very inefficient. For example, if a Filter implementation can determine after examining part of the row that no other pairs in this row will be accepted, there is no mechanism to efficiently skip the remaining Key-Value pairs. Concretely, take a row which is comprised of 1000 Key-Value pairs. After examining the first 10 Key-Value pairs, it is determined that no other Key-Value pairs in this row will be accepted. The Filter must still examine each remaining 990 Key-Value pairs in this row. Another way to express this deficiency is that Filters have no means to leverage the seek method to efficiently skip large portions of Key-Value pairs.

As such, the Filter class functions well for filtering small amounts of data, but is inefficient for filtering large amounts of data. The decision to use a Filter strongly depends on the use case and distribution of data being filtered.

The Combiner class is another common abstract Iterator. Similar to the Combiner interface define in Hadoop’s MapReduce framework, implementations of this abstract class reduce multiple Values for different versions of a Key (Keys which only differ by timestamps) into one Key-Value pair. Combiners provide a simple way to implement common operations like summation and aggregation without the need to implement the entire Accumulo Iterator interface.

One important consideration when choosing to design a Combiner is that the "reduction" operation is often best represented when it is associative and commutative. Operations which do not meet these criteria can be implemented; however, the implementation can be difficult.

A second consideration is that a Combiner is not guaranteed to see every Key-Value pair which differ only by timestamp every time it is invoked. For example, if there are 5 Key-Value pairs in a table which only differ by the timestamps 1, 2, 3, 4, and 5, it is not guaranteed that every invocation of the Combiner will see 5 timestamps. One invocation might see the Values for Keys with timestamp 1 and 4, while another invocation might see the Values for Keys with the timestamps 1, 2, 4 and 5.

Finally, when configuring an Accumulo table to use a Combiner, be sure to disable the Versioning Iterator or set the Combiner at a priority less than the Combiner (the Versioning Iterator is added at a priority of 20 by default). The Versioning Iterator will filter out multiple Key-Value pairs that differ only by timestamp and return only the Key-Value pair that has the largest timestamp.

Commonly, granular Ranges that a client passes to an Iterator from a Scanner or BatchScanner are unmodified. If a Range falls within the boundaries of a Tablet, an Iterator will often see that same Range in the seek method. However, there is no guarantee that the Range will remain unaltered from client to server. As such, Iterators should never make assumptions about the current state/context based on the Range.

The common failure condition is referred to as a "re-seek". In the context of a Scan, TabletServers construct the "stack" of Iterators and batch up Key-Value pairs to send back to the client. When a sufficient number of Key-Value pairs are collected, it is common for the Iterators to be "torn down" until the client asks for the next batch of Key-Value pairs. This is done by the TabletServer to add fairness in ensuring one Scan does not monopolize the available resources. When the client asks for the next batch, the implementation modifies the original Range so that servers know the point to resume the iteration (to avoid returning duplicate Key-Value pairs). Specifically, the new Range is created from the original but is shortened by setting the startKey of the original Range to the Key last returned by the Scan, non-inclusive.

The ability for an Iterator to "skip over" large blocks of Key-Value pairs is a major tenet behind Iterators. By seek'ing when it is known that there is a collection of Key-Value pairs which can be ignored can greatly increase the speed of a scan as many Key-Value pairs do not have to be deserialized and processed.

While the seek method provides the Range that should be used to seek the underlying source Iterator, there is no guarantee that the implementing Iterator uses that Range to perform the seek on its "source" Iterator. As such, it is possible to seek to any Range and the interface has no assertions to prevent this from happening.

Since Iterators are allowed to seek to arbitrary Keys, it also allows Iterators to create infinite loops inside Scans that will repeatedly read the same data without end. If an arbitrary Range is constructed, it should construct a completely new Range as it allows for bugs to be introduced which will break Accumulo.

Thus, seek's should always be thought of as making "forward progress" in the view of the total iteration. The startKey of a Range should always be greater than the current Key seen by the Iterator while the endKey of the Range should always retain the original endKey (and endKey inclusivity) of the last Range seen by your Iterator’s implementation of seek.

Implementations of Iterator might be tempted to open BatchWriters inside of an Iterator as a means to implement triggers for writing additional data outside of their client application. The lifecycle of an Iterator is not managed in such a way that guarantees that this is safe nor efficient. Specifically, there is no way to guarantee that the internal ThreadPool inside of the BatchWriter is closed (and the thread(s) are reaped) without calling the close() method. close'ing and recreating a BatchWriter after every Key-Value pair is also prohibitively performance limiting to be considered an option.

The only safe way to generate additional data in an Iterator is to alter the current Key-Value pair. For example, the WholeRowIterator serializes the all of the Key-Values pairs that fall within each row. A safe way to generate more data in an Iterator would be to construct an Iterator that is "higher" (at a larger priority) than the WholeRowIterator, that is, the Iterator receives the Key-Value pairs which are a serialization of many Key-Value pairs. The custom Iterator could deserialize the pairs, compute some function, and add a new Key-Value pair to the original collection, re-serializing the collection of Key-Value pairs back into a single Key-Value pair.

Any other situation is likely not guaranteed to ensure that the caller (a Scan or a Compaction) will always see all intended data that is generated.

On an instance of an Iterator: init is always called before seek, seek is always called before hasTop, getTopKey and getTopValue will not be called if hasTop returns false.

As mentioned, instance of Iterators may be torn down inside of the server transparently. When a complex collection of iterators is performing some advanced functionality, they will not be torn down until a Key-Value pair is returned out of the "stack" of Iterators (and added into the batch of Key-Values to be returned to the caller). Being torn-down is equivalent to a new instance of the Iterator being creating and deepCopy being called on the new instance with the old instance provided as the argument to deepCopy. References to the old instance are removed and the object is lazily garbage collected by the JVM.

To read from an Accumulo table create a Mapper with the following class parameterization and be sure to configure the AccumuloInputFormat.

To write to an Accumulo table, create a Reducer with the following class parameterization and be sure to configure the AccumuloOutputFormat. The key emitted from the Reducer identifies the table to which the mutation is sent. This allows a single Reducer to write to more than one table if desired. A default table can be configured using the AccumuloOutputFormat, in which case the output table name does not have to be passed to the Context object within the Reducer.

The Text object passed as the output should contain the name of the table to which this mutation should be applied. The Text can be null in which case the mutation will be applied to the default table name specified in the AccumuloOutputFormat options.

Optional Settings:

To restrict Accumulo to a set of row ranges:

To restrict Accumulo to a list of columns:

To use a regular expression to match row IDs:

The AccumuloMultiTableInputFormat allows the scanning over multiple tables in a single MapReduce job. Separate ranges, columns, and iterators can be used for each table.

To set the configuration objects on the job:

Optional settings:

To restrict to a set of ranges:

To restrict Accumulo to a list of columns:

To set scan iterators:

The name of the table can be retrieved from the input split:

Optional Settings:

An example of using MapReduce with Accumulo can be found at accumulo/docs/examples/README.mapred.

An interesting use of combining iterators within an 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 an Accumulo table can be used to efficiently store sparse features and their weights to be incrementally updated via the use of an combining iterator.

Each system involved in replication (even the primary) needs a name that uniquely identifies it across all peers in the replication graph. This should be considered fixed for an instance, and set in accumulo-site.xml.

For each peer of this system, Accumulo needs to know the name of that peer, the class used to replicate data to that system and some configuration information to connect to this remote peer. In the case of Accumulo, this additional data is the Accumulo instance name and ZooKeeper quorum; however, this varies on the replication implementation for the peer.

These can be set in the site configuration to ease deployments; however, as they may change, it can be useful to set this information using the Accumulo shell.

To configure a peer with the name peer1 which is an Accumulo system with an instance name of accumulo_peer and a ZooKeeper quorum of 10.0.0.1,10.0.2.1,10.0.3.1, invoke the following command in the shell.

Since this is an Accumulo system, we also want to set a username and password to use when authenticating with this peer. On our peer, we make a special user which has permission to write to the tables we want to replicate data into, "replication" with a password of "password". We then need to record this in the primary’s configuration.

Alternatively, when configuring replication on Accumulo running Kerberos, a keytab file per peer can be configured instead of a password. The provided keytabs must be readable by the unix user running Accumulo. They keytab for a peer can be unique from the keytab used by Accumulo or any keytabs for other peers.

Now, we presently have a peer defined, so we just need to configure which tables will replicate to that peer. We also need to configure an identifier to determine where this data will be replicated on the peer. Since we’re replicating to another Accumulo cluster, this is a table ID. In this example, we want to enable replication on my_table and configure our peer accumulo_peer as a target, sending the data to the table with an ID of 2 in accumulo_peer.

To replicate a single table on the primary to multiple peers, the second command in the above shell snippet can be issued, for each peer and remote identifier pair.

The AccumuloReplicaSystem uses Thrift to communicate with a peer Accumulo instance and replicate the necessary data. The TabletServer running on the primary will communicate with the Master on the peer to request the address of a TabletServer on the peer which this TabletServer will use to replicate the data.

The TabletServer on the primary will then replicate data in batches of a configurable size (replication.max.unit.size). The TabletServer on the peer will report how many records were applied back to the primary, which will be used to record how many records were successfully replicated. The TabletServer on the primary will continue to replicate data in these batches until no more data can be read from the file.

We can assign the "unique" name that identifies this Accumulo instance among all others that might participate in replication together. In this example, we will use the names provided in the description.

Be sure to use non-local IP addresses. Other nodes need to connect to it and using localhost will likely result in a local node talking to another local node.

The rest of the configuration is dynamic and is best configured on the fly (in ZooKeeper) than in accumulo-site.xml.

The next series of command are to be run on the peer system. Create a user account for the primary instance called "peer". The password for this account will need to be saved in the configuration on the primary

Remember what the table ID for my_table is. You’ll need that to configured the primary instance.

Next, configure the primary instance.

As previously mentioned, the replication feature uses the Write-Ahead Log files for a number of reasons, one of which is to prevent the need for data to be written to RFiles before it is available to be replicated. While this can help reduce the latency for a batch of Mutations that have been written to Accumulo, the latency is at least seconds to tens of seconds for replication once ingest is active. For a table which replication has just been enabled on, this is likely to take a few minutes before replication will begin.

Once ingest is active and flowing into the system at a regular rate, replication should be occurring at a similar rate, given sufficient computing resources. Replication attempts to copy data at a rate that is to be considered low latency but is not a replacement for custom indexing code which can ensure near real-time referential integrity on secondary indexes.

Accumulo Iterators tend to be a heavy hammer which can be used to solve a variety of problems. In general, it is highly recommended that Iterators which are applied at major compaction time are both idempotent and associative due to the non-determinism in which some set of files for a Tablet might be compacted. In practice, this translates to common patterns, such as aggregation, which are implemented in a manner resilient to duplication (such as using a Set instead of a List).

Due to the asynchronous nature of replication and the expectation that hardware failures and network partitions will exist, it is generally not recommended to not configure replication on a table which has Iterators set which are not idempotent. While the replication implementation can make some simple assertions to try to avoid re-replication of data, it is not presently guaranteed that all data will only be sent to a peer once. Data will be replicated at least once. Typically, this is not a problem as the VersioningIterator will automaticaly deduplicate this over-replication because they will have the same timestamp; however, certain Combiners may result in inaccurate aggregations.

As a concrete example, consider a table which has the SummingCombiner configured to sum all values for multiple versions of the same Key. For some key, consider a set of numeric values that are written to a table on the primary: [1, 2, 3]. On the primary, all of these are successfully written and thus the current value for the given key would be 6, (1 + 2 + 3). Consider, however, that each of these updates to the peer were done independently (because other data was also included in the write-ahead log that needed to be replicated). The update with a value of 1 was successfully replicated, and then we attempted to replicate the update with a value of 2 but the remote server never responded. The primary does not know whether the update with a value of 2 was actually applied or not, so the only recourse is to re-send the update. After we receive confirmation that the update with a value of 2 was replicated, we will then replicate the update with 3. If the peer did never apply the first update of 2, the summation is accurate. If the update was applied but the acknowledgement was lost for some reason (system failure, network partition), the update will be resent to the peer. Because addition is non-idempotent, we have created an inconsistency between the primary and peer. As such, the SummingCombiner wouldn’t be recommended on a table being replicated.

While there are changes that could be made to the replication implementation which could attempt to mitigate this risk, presently, it is not recommended to configure Iterators or Combiners which are not idempotent to support cases where inaccuracy of aggregations is not acceptable.

In Accumulo, when more than one key exists that are exactly the same, keys that are equal down to the timestamp, the retained value is non-deterministic. Replication introduces another level of non-determinism in this case. For a table that is being replicated and has multiple equal keys with different values inserted into it, the final value in that table on the primary instance is not guaranteed to be the final value on all replicas.

For example, say the values that were inserted on the primary instance were value1 and value2 and the final value was value1, it is not guaranteed that all replicas will have value1 like the primary. The final value is non-deterministic for each instance.

As is the recommendation without replication enabled, if multiple values for the same key (sans timestamp) are written to Accumulo, it is strongly recommended that the value in the timestamp properly reflects the intended version by the client. That is to say, newer values inserted into the table should have larger timestamps. If the time between writing updates to the same key is significant (order minutes), this concern can likely be ignored.

Currently, files that are bulk imported into a table configured for replication are not replicated. There is no technical reason why it was not implemented, it was simply omitted from the initial implementation. This is considered a fair limitation because bulk importing generated files multiple locations is much simpler than bifurcating "live" ingest data into two instances. Given some existing bulk import process which creates files and them imports them into an Accumulo instance, it is trivial to copy those files to a new HDFS instance and import them into another Accumulo instance using the same process. Hadoop’s distcp command provides an easy way to copy large amounts of data to another HDFS instance which makes the problem of duplicating bulk imports very easy to solve.

This section is for the tracking of a WAL file that needs to be replicated to one or more Accumulo remote tables. This entry is tracking that replication needs to happen on the given WAL file, but also that the local Accumulo table, as specified by the column qualifier "local table ID", has information in this WAL file.

The structure of the key-value is as follows:

This entry is created based on a replication entry from the Accumlo metadata table, and is deleted from the replication table when the WAL has been fully replicated to all remote Accumulo tables.

This section is for the tracking of a WAL file that needs to be replicated to a single Accumulo table in a remote Accumulo cluster. If a WAL must be replicated to multiple tables, there will be multiple entries. The Value for this Key is a serialized ProtocolBuffer message which encapsulates the portion of the WAL which was already sent for this file. The "replication target" is the unique location of where the file needs to be replicated: the identifier for the remote Accumulo cluster and the table ID in that remote Accumulo cluster. The protocol buffer in the value tracks the progress of replication to the remote cluster.

The "work" entry is created when a WAL has an "order" entry, and deleted after the WAL is replicated to all necessary remote clusters.

This section is used to order and schedule (create) replication work. In some cases, data with the same timestamp may be provided multiple times. In this case, it is important that WALs are replicated in the same order they were created/used. In this case (and in cases where this is not important), the order entry ensures that oldest WALs are processed most quickly and pushed through the replication framework.

The "order" entry is created when the WAL is closed (no longer being written to) and is removed when the WAL is fully replicated to all remote locations.

Without any additional arguments, this command will print all operations that still exist in the FATE store (ZooKeeper). This will include active, pending, and completed operations (completed operations are lazily removed from the store). Each operation includes a unique "transaction ID", the state of the operation (e.g. NEW, IN_PROGRESS, FAILED), any locks the transaction actively holds and any locks it is waiting to acquire.

This option can also accept transaction IDs which will restrict the list of transactions shown.

This command can be used to manually fail a FATE transaction and requires a transaction ID as an argument. Failing an operation is not a normal procedure and should only be performed by an administrator who understands the implications of why they are failing the operation.

This command requires a transaction ID and will delete any locks that the transaction holds. Like the fail command, this command should only be used in extreme circumstances by an administrator that understands the implications of the command they are about to invoke. It is not normal to invoke this command.

This command accepts zero more transaction IDs. If given no transaction IDs, it will dump all active transactions. A FATE operations is compromised as a sequence of REPOs. In order to start a FATE transaction, a REPO is pushed onto a per transaction REPO stack. The top of the stack always contains the next REPO the FATE transaction should execute. When a REPO is successful it may return another REPO which is pushed on the stack. The dump command will print all of the REPOs on each transactions stack. The REPOs are serialized to JSON in order to make them human readable.

The truststore.jks file is the Java keystore which contains the certificate authority’s public key.

It’s common that each host in the instance is issued its own certificate (notably to ensure that revocation procedures can be easily followed). The following steps can be taken for each host.

The server.jks file is the Java keystore containing the certificate for a given host. The above methods are equivalent whether the certficate is generate for an Accumulo server or a client.

The first step is to obtain a Kerberos identity for the Accumulo server processes. When running Accumulo with Kerberos enabled, a valid Kerberos identity will be required to initiate any RPC between Accumulo processes (e.g. Master and TabletServer) in addition to any HDFS action (e.g. client to HDFS or TabletServer to HDFS).

Q: I have valid Kerberos credentials and a correct client configuration file but I still get errors like:

A: When you have a valid client configuration and Kerberos TGT, it is possible that the search path for your local credentials cache is incorrect. Check the value of the KRB5CCNAME environment value, and ensure it matches the value reported by klist.

Q: I thought I had everything configured correctly, but my client/server still fails to log in. I don’t know what is actually failing.

A: Add the following system property to the JVM invocation:

This will enable lots of extra debugging at the JVM level which is often sufficient to diagnose some high-level configuration problem. Client applications can add this system property by hand to the command line and Accumulo server processes or applications started using the accumulo script by adding the property to ACCUMULO_GENERAL_OPTS in $ACCUMULO_CONF_DIR/accumulo-env.sh.

Additionally, you can increase the log4j levels on org.apache.hadoop.security, which includes the Hadoop UserGroupInformation class, which will include some high-level debug statements. This can be controlled in your client application, or using $ACCUMULO_CONF_DIR/generic_logger.xml

Q: All of my Accumulo processes successfully start and log in with their keytab, but they are unable to communicate with each other, showing the following errors:

or

A: As previously mentioned, the hostname, and subsequently the address each Accumulo process is bound/listening on, is extremely important when negotiating an SASL connection. This problem commonly arises when the Accumulo servers are not configured to listen on the address denoted by their FQDN.

The values in the Accumulo "hosts" files (In $ACCUMULO_CONF_DIR: masters, monitors, slaves, tracers, and gc) should match the instance componentof the Kerberos server principal (e.g. host in accumulo/host@EXAMPLE.COM).

Q: After configuring my system for Kerberos, server processes come up normally and I can interact with the system. However, when I attempt to use the "Recent Traces" page on the Monitor UI I get a stacktrace similar to:

A: This indicates that the Monitor has not been able to successfully log in a client-side user to read from the trace table. Accumulo allows the TraceServer to rely on the property general.kerberos.keytab as a fallback when logging in the trace user if the trace.token.property.keytab property isn’t defined. Some earlier versions of Accumulo did not do this same fallback for the Monitor’s use of the trace user. The end result is that if you configure general.kerberos.keytab and not trace.token.property.keytab you will end up with a system that properly logs trace information but can’t view it.

Ensure you have set trace.token.property.keytab to point to a keytab for the principal defined in trace.user in the accumulo-site.xml file for the Monitor, since that should work in all versions of Accumulo.

Accumulo needs to know where to find the software it depends on. Edit accumulo-env.sh and specify the following:

By default Accumulo TabletServers are set to use 1GB of memory. You may change this by altering the value of $ACCUMULO_TSERVER_OPTS. Note the syntax is that of the Java JVM command line options. This value should be less than the physical memory of the machines running TabletServers.

There are similar options for the master’s memory usage and the garbage collector process. Reduce these if they exceed the physical RAM of your hardware and increase them, within the bounds of the physical RAM, if a process fails because of insufficient memory.

Note that you will be specifying the Java heap space in accumulo-env.sh. You should make sure that the total heap space used for the Accumulo tserver and the Hadoop DataNode and TaskTracker is less than the available memory on each slave node in the cluster. On large clusters, it is recommended that the Accumulo master, Hadoop NameNode, secondary NameNode, and Hadoop JobTracker all be run on separate machines to allow them to use more heap space. If you are running these on the same machine on a small cluster, likewise make sure their heap space settings fit within the available memory.

The tablet server uses a data structure called a MemTable to store sorted key/value pairs in memory when they are first received from the client. When a minor compaction occurs, this data structure is written to HDFS. The MemTable will default to using memory in the JVM but a JNI version, called the native map, can be used to significantly speed up performance by utilizing the memory space of the native operating system. The native map also avoids the performance implications brought on by garbage collection in the JVM by causing it to pause much less frequently.

On the machine that will serve as the Accumulo master:

Note that if using domain names rather than IP addresses, DNS must be configured properly for all machines participating in the cluster. DNS can be a confusing source of errors.

Specify appropriate values for the following settings in $ACCUMULO_HOME/conf/accumulo-site.xml :

This enables Accumulo to find ZooKeeper. Accumulo uses ZooKeeper to coordinate settings between processes and helps finalize TabletServer failure.

The instance needs a secret to enable secure communication between servers. Configure your secret and make sure that the accumulo-site.xml file is not readable to other users. For alternatives to storing the instance.secret in plaintext, please read the Sensitive Configuration Values section.

Some settings can be modified via the Accumulo shell and take effect immediately, but some settings require a process restart to take effect. See the configuration documentation (available in the docs directory of the tarball and in Configuration Management) for details.

One aspect of Accumulo’s configuration which is different as compared to the rest of the Hadoop ecosystem is that the server-process classpath is determined in part by multiple values. A bootstrap classpath is based soley on the accumulo-start.jar, Log4j and $ACCUMULO_CONF_DIR.

A second classloader is used to dynamically load all of the resources specified by general.classpaths in $ACCUMULO_CONF_DIR/accumulo-site.xml. This value is a comma-separated list of regular-expression paths which are all loaded into a secondary classloader. This includes Hadoop, Accumulo and ZooKeeper jars necessary to run Accumulo. When this value is not defined, a default value is used which attempts to load Hadoop from multiple potential locations depending on how Hadoop was installed. It is strongly recommended that general.classpaths is defined and limited to only the necessary jars to prevent extra jars from being unintentionally loaded into Accumulo processes.

Accumulo has a number of configuration files which can contain references to other hosts in your network. All of the "host" configuration files for Accumulo (gc, masters, slaves, monitor, tracers) as well as instance.volumes in accumulo-site.xml must contain some host reference.

While IP address, short hostnames, or fully qualified domain names (FQDN) are all technically valid, it is good practice to always use FQDNs for both Accumulo and other processes in your Hadoop cluster. Failing to consistently use FQDNs can have unexpected consequences in how Accumulo uses the FileSystem.

A common way for this problem can be observed is via applications that use Bulk Ingest. The Accumulo Master coordinates moving the input files to Bulk Ingest to an Accumulo-managed directory. However, Accumulo cannot safely move files across different Hadoop FileSystems. This is problematic because Accumulo also cannot make reliable assertions across what is the same FileSystem which is specified with different names. Naively, while 127.0.0.1:8020 might be a valid identifier for an HDFS instance, Accumulo identifies localhost:8020 as a different HDFS instance than 127.0.0.1:8020.

Copy the masters, slaves, accumulo-env.sh, and if necessary, accumulo-site.xml from the $ACCUMULO_HOME/conf/ directory on the master to all the machines specified in the slaves file.

Accumulo has a number of properties that can be specified via the accumulo-site.xml file which are sensitive in nature, instance.secret and trace.token.property.password are two common examples. Both of these properties, if compromised, have the ability to result in data being leaked to users who should not have access to that data.

In Hadoop-2.6.0, a new CredentialProvider class was introduced which serves as a common implementation to abstract away the storage and retrieval of passwords from plaintext storage in configuration files. Any Property marked with the Sensitive annotation is a candidate for use with these CredentialProviders. For version of Hadoop which lack these classes, the feature will just be unavailable for use.

A comma separated list of CredentialProviders can be configured using the Accumulo Property general.security.credential.provider.paths. Each configured URL will be consulted when the Configuration object for accumulo-site.xml is accessed.

One of the implementations provided in Hadoop-2.6.0 is a Java KeyStore CredentialProvider. Each entry in the KeyStore is the Accumulo Property key name. For example, to store the instance.secret, the following command can be used:

The command will then prompt you to enter the secret to use and create a keystore in:

Then, accumulo-site.xml must be configured to use this KeyStore as a CredentialProvider:

This configuration will then transparently extract the instance.secret from the configured KeyStore and alleviates a human readable storage of the sensitive property.

A KeyStore can also be stored in HDFS, which will make the KeyStore readily available to all Accumulo servers. If the local filesystem is used, be aware that each Accumulo server will expect the KeyStore in the same location.

In version 1.6.0, Accumulo included a new type of configuration file known as a client configuration file. One problem with the traditional "site.xml" file that is prevalent through Hadoop is that it is a single file used by both clients and servers. This makes it very difficult to protect secrets that are only meant for the server processes while allowing the clients to connect to the servers.

The client configuration file is a subset of the information stored in accumulo-site.xml meant only for consumption by clients of Accumulo. By default, Accumulo checks a number of locations for a client configuration by default:

These files are Java Properties files. These files can currently contain information about ZooKeeper servers, RPC properties (such as SSL or SASL connectors), distributed tracing properties. Valid properties are defined by the ClientProperty enum contained in the client API.

Accumulo has the ability for users to add custom tags to tables. This allows applications to set application-level metadata about a table. These tags can be anything from a table description, administrator notes, date created, etc. This is done by naming and setting a property with a prefix table.custom.*.

Currently, table properties are stored in ZooKeeper. This means that the number and size of custom properties should be restricted on the order of 10’s of properties at most without any properties exceeding 1MB in size. ZooKeeper’s performance can be very sensitive to an excessive number of nodes and the sizes of the nodes. Applications which leverage the user of custom properties should take these warnings into consideration. There is no enforcement of these warnings via the API.

Accumulo loads classes from the locations specified in the general.classpaths property. Additionally, Accumulo will load classes from the locations specified in the general.dynamic.classpaths property and will monitor and reload them if they change. The reloading feature is useful during the development and testing of iterators as new or modified iterator classes can be deployed to Accumulo without having to restart the database.

Accumulo also has an alternate configuration for the classloader which will allow it to load classes from remote locations. This mechanism uses Apache Commons VFS which enables locations such as http and hdfs to be used. This alternate configuration also uses the general.classpaths property in the same manner described above. It differs in that you need to configure the general.vfs.classpaths property instead of the general.dynamic.classpath property. As in the default configuration, this alternate configuration will also monitor the vfs locations for changes and reload if necessary.

Make sure Hadoop is configured on all of the machines in the cluster, including access to a shared HDFS instance. Make sure HDFS and ZooKeeper are running. Make sure ZooKeeper is configured and running on at least one machine in the cluster. Start Accumulo using the bin/start-all.sh script.

To verify that Accumulo is running, check the Status page as described in Monitoring. In addition, the Shell can provide some information about the status of tables via reading the metadata tables.

To shutdown cleanly, run bin/stop-all.sh and the master will orchestrate the shutdown of all the tablet servers. Shutdown waits for all minor compactions to finish, so it may take some time for particular configurations.

Update your $ACCUMULO_HOME/conf/slaves (or $ACCUMULO_CONF_DIR/slaves) file to account for the addition.

Next, ssh to each of the hosts you want to add and run:

Make sure the host in question has the new configuration, or else the tablet server won’t start; at a minimum this needs to be on the host(s) being added, but in practice it’s good to ensure consistent configuration across all nodes.

If you need to take a node out of operation, you can trigger a graceful shutdown of a tablet server. Accumulo will automatically rebalance the tablets across the available tablet servers.

Alternatively, you can ssh to each of the hosts you want to remove and run:

Be sure to update your $ACCUMULO_HOME/conf/slaves (or $ACCUMULO_CONF_DIR/slaves) file to account for the removal of these hosts. Bear in mind that the monitor will not re-read the slaves file automatically, so it will report the decomissioned servers as down; it’s recommended that you restart the monitor so that the node list is up to date.

Occasionally, it might be necessary to restart the processes on a specific node. In addition to the start-all.sh and stop-all.sh scripts, Accumulo contains scripts to start/stop all processes on a node and start/stop a given process on a node.

start-here.sh and stop-here.sh will start/stop all Accumulo processes on the current node. The necessary processes to start/stop are determined via the "hosts" files (e.g. slaves, masters, etc). These scripts expect no arguments.

start-server.sh can also be useful in starting a given process on a host. The first argument to the process is the hostname of the machine. Use the same host that you specified in hosts file (if you specified FQDN in the masters file, use the FQDN, not the shortname). The second argument is the name of the process to start (e.g. master, tserver).

The steps described to decomission a node can also be used (without removal of the host from the $ACCUMULO_HOME/conf/slaves file) to gracefully stop a node. This will ensure that the tabletserver is cleanly stopped and recovery will not need to be performed when the tablets are re-hosted.

With very powerful nodes, it may be beneficial to run more than one TabletServer on a given node. This decision should be made carefully and with much deliberation as Accumulo is designed to be able to scale to using 10’s of GB of RAM and 10’s of CPU cores.

To run multiple TabletServers on a single host you will need to change the NUM_TSERVERS property in the accumulo-env.sh file from 1 to the number of TabletServers that you want to run. On NUMA hardware, with numactl installed, the TabletServer will interleave its memory allocations across the NUMA nodes and the processes will be scheduled on all the NUMA cores without restriction. To change this behavior you can uncomment the TSERVER_NUMA_OPTIONS example in accumulo-env.sh and set the numactl options for each TabletServer.

Accumulo TabletServers bind certain ports on the host to accommodate remote procedure calls to/from other nodes. Running more than one TabletServer on a host requires that you set the following properties in accumulo-site.xml:

Accumulo’s provided scripts for starting and stopping the cluster should work normally with multiple TabletServers on a host. Sanity checks are provided in the scripts and will output an error when there is a configuration mismatch.

The Accumulo Monitor provides an interface for monitoring the status and health of Accumulo components. The Accumulo Monitor provides a web UI for accessing this information at http://monitorhost:9995/.

Things highlighted in yellow may be in need of attention. If anything is highlighted in red on the monitor page, it is something that definitely needs attention.

The Overview page contains some summary information about the Accumulo instance, including the version, instance name, and instance ID. There is a table labeled Accumulo Master with current status, a table listing the active Zookeeper servers, and graphs displaying various metrics over time. These include ingest and scan performance and other useful measurements.

The Master Server, Tablet Servers, and Tables pages display metrics grouped in different ways (e.g. by tablet server or by table). Metrics typically include number of entries (key/value pairs), ingest and query rates. The number of running scans, major and minor compactions are in the form number_running (number_queued). Another important metric is hold time, which is the amount of time a tablet has been waiting but unable to flush its memory in a minor compaction.

The Server Activity page graphically displays tablet server status, with each server represented as a circle or square. Different metrics may be assigned to the nodes' color and speed of oscillation. The Overall Avg metric is only used on the Server Activity page, and represents the average of all the other metrics (after normalization). Similarly, the Overall Max metric picks the metric with the maximum normalized value.

The Garbage Collector page displays a list of garbage collection cycles, the number of files found of each type (including deletion candidates in use and files actually deleted), and the length of the deletion cycle. The Traces page displays data for recent traces performed (see the following section for information on Tracing). The Recent Logs page displays warning and error logs forwarded to the monitor from all Accumulo processes. Also, the XML and JSON links provide metrics in XML and JSON formats, respectively.

The Accumulo monitor does a best-effort to not display any sensitive information to users; however, the monitor is intended to be a tool used with care. It is not a production-grade webservice. It is a good idea to whitelist access to the monitor via an authentication proxy or firewall. It is strongly recommended that the Monitor is not exposed to any publicly-accessible networks.

SSL may be enabled for the monitor page by setting the following properties in the accumulo-site.xml file:

If the Accumulo conf directory has been configured (in particular the accumulo-env.sh file must be set up), the generate_monitor_certificate.sh script in the Accumulo bin directory can be used to create the keystore and truststore files with random passwords. The script will print out the properties that need to be added to the accumulo-site.xml file. The stores can also be generated manually with the Java keytool command, whose usage can be seen in the generate_monitor_certificate.sh script.

If desired, the SSL ciphers allowed for connections can be controlled via the following properties in accumulo-site.xml:

If SSL is enabled, the monitor URL can only be accessed via https. This also allows you to access the Accumulo shell through the monitor page. The left navigation bar will have a new link to Shell. An Accumulo user name and password must be entered for access to the shell.

Metrics2 is configured by examining the classpath for a file that matches hadoop-metrics2*.properties. The example configuration files that Accumulo provides for use include hadoop-metrics2-accumulo.properties as a template which can be used to enable file, Graphite or Ganglia sinks (some minimal configuration required for Graphite and Ganglia). Because the Hadoop configuration is also on the Accumulo classpath, be sure that you do not have multiple Metrics2 configuration files. It is recommended to consolidate metrics in a single properties file in a central location to remove ambiguity. The contents of hadoop-metrics2-accumulo.properties can be added to a central hadoop-metrics2.properties in $HADOOP_CONF_DIR.

As a note for configuring the file sink, the provided path should be absolute. A relative path or file name will be created relative to the directory in which the Accumulo process was started. External tools, such as logrotate, can be used to prevent these files from growing without bound.

Each server process should have log messages from the Metrics2 library about the sinks that were created. Be sure to check the Accumulo processes log files when debugging missing metrics output.

For additional information on configuring Metrics2, visit the Javadoc page for Metrics2.

To collect traces, Accumulo needs at least one server listed in $ACCUMULO_HOME/conf/tracers. The server collects traces from clients and writes them to the trace table. The Accumulo user that the tracer connects to Accumulo with can be configured with the following properties (see the Configuration section for setting Accumulo server properties)

Other tracer configuration properties include

The zookeeper path is configured to /tracers by default. If multiple Accumulo instances are sharing the same ZooKeeper quorum, take care to configure Accumulo with unique values for this property.

Traces are collected via SpanReceivers. The default SpanReceiver configured is org.apache.accumulo.core.trace.ZooTraceClient, which sends spans to an Accumulo Tracer process, as discussed in the previous section. This default can be changed to a different span receiver, or additional span receivers can be added in a comma-separated list, by modifying the property

Individual span receivers may require their own configuration parameters, which are grouped under the trace.span.receiver.* prefix. ZooTraceClient uses the following properties. The first three properties are populated from other Accumulo properties, while the remaining ones should be prefixed with trace.span.receiver. when set in the Accumulo configuration.

Note that to configure an Accumulo client for tracing, including the Accumulo shell, the client configuration must be given the same trace.span.receivers, trace.span.receiver.*, and trace.zookeeper.path properties as the servers have.

Hadoop can also be configured to send traces to Accumulo, as of Hadoop 2.6.0, by setting properties in Hadoop’s core-site.xml file. Instead of using the trace.span.receiver.* prefix, Hadoop uses hadoop.htrace.*. The Hadoop configuration does not have access to Accumulo’s properties, so the hadoop.htrace.tracer.zookeeper.host property must be specified. The zookeeper timeout defaults to 30000 (30 seconds), and the zookeeper path defaults to /tracers. An example of configuring Hadoop to send traces to ZooTraceClient is

The accumulo-core, accumulo-tracer, accumulo-fate and libthrift jars must also be placed on Hadoop’s classpath.

Tracing can be used to measure a client operation, such as a scan, as the operation traverses the distributed system. To enable tracing for your application call

Once tracing has been enabled, a client can wrap an operation in a trace.

The user can create additional Spans within a Trace.

The sampler (such as Sampler.ALWAYS) for the trace should only be specified with a top-level span, and subsequent spans will be collected depending on whether that first span was sampled. Don’t forget to specify a Sampler at the top-level span because the default Sampler only samples when part of a pre-existing trace, which will never occur in a client that never specifies a Sampler.

Like Dapper, Accumulo tracing supports user defined annotations to associate additional data with a Trace. Checking whether currently tracing is necessary when using a sampler other than Sampler.ALWAYS.

It is also possible to add timeline annotations to your spans. This associates a string with a given timestamp between the start and stop times for a span.

Some client operations may have a high volume within your application. As such, you may wish to only sample a percentage of operations for tracing. As seen below, the CountSampler can be used to help enable tracing for 1-in-1000 operations

Remember to close all spans and disable tracing when finished.

To view collected traces, use the "Recent Traces" link on the Monitor UI. You can also programmatically access and print traces using the TraceDump class.

You can enable tracing for operations run from the shell by using the trace on and trace off commands.

In addition to the general failure modes of running Accumulo, VMs can introduce a couple of environmental challenges that can affect process stability. Clock drift is something that is more common in VMs, especially when VMs are suspended and resumed. Clock drift can cause Accumulo servers to assume that they have lost connectivity to the other Accumulo processes and/or lose their locks in Zookeeper. VM environments also frequently have constrained resources, such as CPU, RAM, network, and disk throughput and capacity. Accumulo generally deals well with constrained resources from a stability perspective (optimizing performance will require additional tuning, which is not covered in this section), however there are some limits.

Each release of Accumulo is built with a specific version of Apache Hadoop, Apache ZooKeeper and Apache Thrift. We expect Accumulo to work with versions that are API compatable with those versions. However this compatibility is not guaranteed because Hadoop, ZooKeeper and Thift may not provide guarantees between their own versions. We have also found that certain versions of Accumulo and Hadoop included bugs that greatly affected overall stability. Thrift is particularly prone to compatablity changes between versions and you must use the same version your Accumulo is built with.

Please check the release notes for your Accumulo version or use the mailing lists at https://accumulo.apache.org for more info.

Q: My tablet server lost its lock. Why?

The primary reason a tablet server loses its lock is that it has been pushed into swap.

A large java program (like the tablet server) may have a large portion of its memory image unused. The operation system will favor pushing this allocated, but unused memory into swap so that the memory can be re-used as a disk buffer. When the java virtual machine decides to access this memory, the OS will begin flushing disk buffers to return that memory to the VM. This can cause the entire process to block long enough for the zookeeper lock to be lost.

A: Configure your system to reduce the kernel parameter swappiness from the default (60) to zero.

Q: My tablet server lost its lock, and I have already set swappiness to zero. Why?

Be careful not to over-subscribe memory. This can be easy to do if your accumulo processes run on the same nodes as hadoop’s map-reduce framework. Remember to add up:

If a 16G node can run 2 mappers and 2 reducers, and each can be 2G, then there is only 8G for the data node, tserver, task tracker and OS.

A: Reduce the memory footprint of each component until it fits comfortably.

Q: My tablet server lost its lock, swappiness is zero, and my node has lots of unused memory!

The JVM memory garbage collector may fall behind and cause a "stop-the-world" garbage collection. On a large memory virtual machine, this collection can take a long time. This happens more frequently when the JVM is getting low on free memory. Check the logs of the tablet server. You will see lines like this:

When freemem becomes small relative to the amount of memory needed, the JVM will spend more time finding free memory than performing work. This can cause long delays in sending keep-alive messages to zookeeper.

A: Ensure the tablet server JVM is not running low on memory.

Q: I’m seeing errors in tablet server logs that include the words "MutationsRejectedException" and "# constraint violations: 1". Moments after that the server died.

The error you are seeing is part of a failing tablet server scenario. This is a bit complicated, so name two of your tablet servers A and B.

Tablet server A is hosting a tablet, let’s call it a-tablet.

Tablet server B is hosting a metadata tablet, let’s call it m-tablet.

m-tablet records the information about a-tablet, for example, the names of the files it is using to store data.

When A ingests some data, it eventually flushes the updates from memory to a file.

Tablet server A then writes this new information to m-tablet, on Tablet server B.

Here’s a likely failure scenario:

Tablet server A does not have enough memory for all the processes running on it. The operating system sees a large chunk of the tablet server being unused, and swaps it out to disk to make room for other processes. Tablet server A does a java memory garbage collection, which causes it to start using all the memory allocated to it. As the server starts pulling data from swap, it runs very slowly. It fails to send the keep-alive messages to zookeeper in a timely fashion, and it looses its zookeeper session.

But, it’s running so slowly, that it takes a moment to realize it should no longer be hosting tablets.

The thread that is flushing a-tablet memory attempts to update m-tablet with the new file information.

Fortunately there’s a constraint on m-tablet. Mutations to the metadata table must contain a valid zookeeper session. This prevents tablet server A from making updates to m-tablet when it no long has the right to host the tablet.

The "MutationsRejectedException" error is from tablet server A making an update to tablet server B’s m-tablet. It’s getting a constraint violation: tablet server A has lost its zookeeper session, and will fail momentarily.

A: Ensure that memory is not over-allocated. Monitor swap usage, or turn swap off.

Q: My accumulo client is getting a MutationsRejectedException. The monitor is displaying "No Such SessionID" errors.

When your client starts sending mutations to accumulo, it creates a session. Once the session is created, mutations are streamed to accumulo, without acknowledgement, against this session. Once the client is done, it will close the session, and get an acknowledgement.

If the client fails to communicate with accumulo, it will release the session, assuming that the client has died. If the client then attempts to send more mutations against the session, you will see "No Such SessionID" errors on the server, and MutationRejectedExceptions in the client.

The client library should be either actively using the connection to the tablet servers, or closing the connection and sessions. If the session times out, something is causing your client to pause.

The most frequent source of these pauses are java garbage collection pauses due to the JVM running out of memory, or being swapped out to disk.

A: Ensure your client has adequate memory and is not being swapped out to disk.

Q: I had disasterous HDFS failure. After bringing everything back up, several tablets refuse to go online.

Data written to tablets is written into memory before being written into indexed files. In case the server is lost before the data is saved into a an indexed file, all data stored in memory is first written into a write-ahead log (WAL). When a tablet is re-assigned to a new tablet server, the write-ahead logs are read to recover any mutations that were in memory when the tablet was last hosted.

If a write-ahead log cannot be read, then the tablet is not re-assigned. All it takes is for one of the blocks in the write-ahead log to be missing. This is unlikely unless multiple data nodes in HDFS have been lost.

A: Get the WAL files online and healthy. Restore any data nodes that may be down.

Q: How do find out which tablets are offline?

A: Use accumulo admin checkTablets

Q: I lost three data nodes, and I’m missing blocks in a WAL. I don’t care about data loss, how can I get those tablets online?

See the discussion in System Metadata Tables, which shows a typical metadata table listing. The entries with a column family of log are references to the WAL for that tablet. If you know what WAL is bad, you can find all the references with a grep in the shell:

A: You can remove the WAL references in the metadata table.

Note: the colon (:) is omitted when specifying the row cf cq for the delete command.

The master will automatically discover the tablet no longer has a bad WAL reference and will assign the tablet. You will need to remove the reference from all the tablets to get them online.

Q: The metadata (or root) table has references to a corrupt WAL.

This is a much more serious state, since losing updates to the metadata table will result in references to old files which may not exist, or lost references to new files, resulting in tablets that cannot be read, or large amounts of data loss.

The best hope is to restore the WAL by fixing HDFS data nodes and bringing the data back online. If this is not possible, the best approach is to re-create the instance and bulk import all files from the old instance into a new tables.

A complete set of instructions for doing this is outside the scope of this guide, but the basic approach is:

Q: One or more HDFS Files under /accumulo/tables are corrupt

Accumulo maintains multiple references into the tablet files in the metadata tables and within the tablet server hosting the file, this makes it difficult to reliably just remove those references.

The directory structure in HDFS for tables will follow the general structure:

If files under /accumulo/tables are corrupt, the best course of action is to recover those files in hdsf see the section on HDFS. Once these recovery efforts have been exhausted, the next step depends on where the missing file(s) are located. Different actions are required when the bad files are in Accumulo data table files or if they are metadata table files.

Data File Corruption

When an Accumulo data file is corrupt, the most reliable way to restore Accumulo operations is to replace the missing file with an “empty” file so that references to the file in the METADATA table and within the tablet server hosting the file can be resolved by Accumulo. An empty file can be created using the CreateEmpty utiity:

The process is to delete the corrupt file and then move the empty file into its place (The generated empty file can be copied and used multiple times if necessary and does not need to be regenerated each time)

Metadata File Corruption

If the corrupt files are metadata files, see System Metadata Tables (under the path /accumulo/tables/!0) then you will need to rebuild the metadata table by initializing a new instance of Accumulo and then importing all of the existing data into the new instance. This is the same procedure as recovering from a zookeeper failure (see ZooKeeper Failure), except that you will have the benefit of having the existing user and table authorizations that are maintained in zookeeper.

You can use the DumpZookeeper utility to save this information for reference before creating the new instance. You will not be able to use RestoreZookeeper because the table names and references are likely to be different between the original and the new instances, but it can serve as a reference.

A: If the files cannot be recovered, replace corrupt data files with a empty rfiles to allow references in the metadata table and in the tablet servers to be resolved. Rebuild the metadata table if the corrupt files are metadata files.

Write-Ahead Log(WAL) File Corruption

In certain versions of Accumulo, a corrupt WAL file (caused by HDFS corruption or a bug in Accumulo that created the file) can block the successful recovery of one to many Tablets. Accumulo can be stuck in a loop trying to recover the WAL file, never being able to succeed.

In the cases where the WAL file’s original contents are unrecoverable or some degree of data loss is acceptable (beware if the WAL file contains updates to the Accumulo metadat table!), the following process can be followed to create an valid, empty WAL file. Run the following commands as the Accumulo unix user (to ensure that the proper file permissions in HDFS)

The above creates a file with the text "--- Log File Header (v2) ---" and then four bytes. You should verify the contents of the file with a hexdump tool.

Then, place this empty WAL in HDFS and then replace the corrupt WAL file in HDFS with the empty WAL.

After the corrupt WAL file has been replaced, the system should automatically recover. It may be necessary to restart the Accumulo Master process as an exponential backup policy is used which could lead to a long wait before Accumulo will try to re-load the WAL file.

Q: I lost my ZooKeeper quorum (hardware failure), but HDFS is still intact. How can I recover my Accumulo instance?

ZooKeeper, in addition to its lock-service capabilities, also serves to bootstrap an Accumulo instance from some location in HDFS. It contains the pointers to the root tablet in HDFS which is then used to load the Accumulo metadata tablets, which then loads all user tables. ZooKeeper also stores all namespace and table configuration, the user database, the mapping of table IDs to table names, and more across Accumulo restarts.

Presently, the only way to recover such an instance is to initialize a new instance and import all of the old data into the new instance. The easiest way to tackle this problem is to first recreate the mapping of table ID to table name and then recreate each of those tables in the new instance. Set any necessary configuration on the new tables and add some split points to the tables to close the gap between how many splits the old table had and no splits.

The directory structure in HDFS for tables will follow the general structure:

For each table, make a new directory that you can move (or copy if you have the HDFS space to do so) all of the rfiles for a given table into. For example, to process the table with an ID of 1, make a new directory, say /new-table-1 and then copy all files from /accumulo/tables/1/*/*.rf into that directory. Additionally, make a directory, /new-table-1-failures, for any failures during the import process. Then, issue the import command using the Accumulo shell into the new table, telling Accumulo to not re-set the timestamp:

Any RFiles which were failed to be loaded will be placed in /new-table-1-failures. Rfiles that were successfully imported will no longer exist in /new-table-1. For failures, move them back to the import directory and retry the importdirectory command.

It is extremely important to note that this approach may introduce stale data back into the tables. For a few reasons, RFiles may exist in the table directory which are candidates for deletion but have not yet been deleted. Additionally, deleted data which was not compacted away, but still exists in write-ahead logs if the original instance was somehow recoverable, will be re-introduced in the new instance. Table splits and merges (which also include the deleteRows API call on TableOperations, are also vulnerable to this problem. This process should not be used if these are unacceptable risks. It is possible to try to re-create a view of the accumulo.metadata table to prune out files that are candidates for deletion, but this is a difficult task that also may not be entirely accurate.

Likewise, it is also possible that data loss may occur from write-ahead log (WAL) files which existed on the old table but were not minor-compacted into an RFile. Again, it may be possible to reconstruct the state of these WAL files to replay data not yet in an RFile; however, this is a difficult task and is not implemented in any automated fashion.

A: The importdirectory shell command can be used to import RFiles from the old instance into a newly created instance, but extreme care should go into the decision to do this as it may result in reintroduction of stale data or the omission of new data.

Table properties are applied to the entire cluster when set in zookeeper using the accumulo API or shell. While table properties take precedent over system properties, both will override properties set in accumulo-site.xml

Table properties consist of all properties with the table.* prefix. Table properties are configured on a per-table basis using the following shell commmand:

System properties are applied to the entire cluster when set in zookeeper using the accumulo API or shell. System properties consist of all properties with a yes in the Zookeeper Mutable column in the table below. They are set with the following shell command:

If a table.* property is set using this method, the value will apply to all tables except those configured on per-table basis (which have higher precedence).

While most system properties take effect immediately, some require a restart of the process which is indicated in Zookeeper Mutable.

Accumulo processes (master, tserver, etc) read their local accumulo-site.xml on start up. Therefore, changes made to accumulo-site.xml must rsynced across the cluster and processes must be restarted to apply changes.

Certain properties (indicated by a no in Zookeeper Mutable) cannot be set in zookeeper and only set in this file. The accumulo-site.xml also allows you to configure tablet servers with different settings.

All properties have a default value in the source code. This value has the lowest precedence and is overriden if set in accumulo-site.xml or zookeeper.

While the default value is usually optimal, there are cases where a change can increase query and ingest performance.

Any properties that are stored in ZooKeeper should consider the limitations of ZooKeeper itself with respect to the number of nodes and the size of the node data. Custom table properties and options for Iterators configured on tables are two areas in which there aren’t any failsafes built into the API that can prevent the user from making this mistake.

While these properties have the ability to add some much needed dynamic configuration tools, use cases which might fall into these warnings should be reconsidered.

Properties in this category related to the configuration of SSL keys for RPC. See also instance.ssl.enabled

Properties in this category must be consistent throughout a cloud. This is enforced and servers won’t be able to communicate if these differ.

Deprecated. Prefix that allows configuration of users that are allowed to impersonate other users

Properties in this category affect the behavior of accumulo overall, but do not have to be consistent throughout a cloud.

Properties in this category affect the behavior of the master server

Properties in this category affect the behavior of the tablet servers

Allows configuration of implementation used to apply replicated data

Properties in this category affect the behavior of the accumulo garbage collector.

Properties in this category affect the behavior of the monitor web server.

Properties in this category affect the behavior of distributed tracing.

Prefix for span receiver configuration properties

The prefix used to create a token for storing distributed traces. For each propetry required by trace.token.type, place this prefix in front of it.

Properties in this category affect tablet server treatment of tablets, but can be configured on a per-table basis. Setting these properties in the site file will override the default globally for all tables and not any specific table. However, both the default and the global setting can be overridden per table using the table operations API or in the shell, which sets the overridden value in zookeeper. Restarting accumulo tablet servers after setting these properties in the site file will cause the global setting to take effect. However, you must use the API or the shell to change properties in zookeeper that are set on a table.