BatchWriters

High throughput when the BatchWriter is used depends on the ability to amortize network overhead by grouping mutations together before shipping them to tablet servers. The BatchWriter does this by waiting for several mutations to become available so it can group them together and send them all at once. This way, it only has to pay the network overhead for every hundred or thousand or so mutations, rather than every mutation. Seen another way, the network cost of writing a mutation can be reduced by a hundred or a thousand times via batching.

The time the BatchWriter spends waiting to gather some number of mutations contributes to latency. If the BatchWriter is able to gather 1,000 mutations, throughput will be higher, but latency will also be higher. Applications can choose therefore to configure the BatchWriter to send smaller batches of mutations more frequently so that latency is kept low. This configuration allows applications to choose where they want to reside in the spectrum between high throughput and low latency. Applications could choose to optimize for low latency at the cost of throughput by configuring the BatchWriter to wait for only a short time to gather a batch of mutations or by explicitly calling flush() to immediately send pending mutations. These settings are detailed in “Committing Mutations”.

Bulk Loading

An alternative to ingesting data via clients using the BatchWriter is to prepare key-value pairs into files and bulk-load them to Accumulo tables. Using bulk loading to write data to Accumulo tables represents sliding all the way to the high-throughput end of the spectrum. Bulk loading employs the MapReduce framework, which is optimized to process data at very high throughput. The data is only available for query after all the data has been processed and the MapReduce job completes, representing higher latency.

To understand how bulk loading works, it’s helpful to know how data for Accumulo tables is stored in HDFS. An Accumulo table consists of a set of files in HDFS and metadata describing which files belong to which tablets and which key ranges each tablet spans. When data is ingested via an Accumulo client, RFiles are created and stored in HDFS as part of the minor compaction process. After tablets are split, they can share RFiles until a major compaction process writes out a separate set of RFiles for each tablet.

Eventually, Accumulo will end up having one or a small number of RFiles for each tablet, and each RFile will only contain data for one tablet. This represents a kind of equilibrium state for Accumulo in which no more compactions are necessary.

Users who want to get data into Accumulo at a very high rate of throughput can use MapReduce to create the RFiles such that they closely resemble the set of RFiles that Accumulo would create on its own if the data were to be ingested via streaming clients.

Creating RFiles via MapReduce can be faster because a data set can be organized into the optimal set of RFiles in one MapReduce job rather than via several rounds of compaction on intermediate RFiles. The downside of bulk loading is that none of the data is available for query until the entire data set is done being processed by MapReduce. It also requires that all the data to be loaded is staged in HDFS.

Bulk loading is an option for quickly loading a large data set into Accumulo when it is possible to stage the data in HDFS and when the latency requirements are such that the data can be unavailable until the MapReduce job is complete.

“Bulk-loading files from a MapReduce job” discusses additional factors to consider when bulk loading, including how to handle key timestamps.

Storage Devices

Unlike some databases, Accumulo is designed to keep most of the data managed on disk. As much data as will fit is cached into RAM as data is read from disk, but even reads that request data that is not cached in RAM are designed to be fast, because Accumulo minimizes disk seeks by keeping the data organized and reading fairly large chunks at a time.

Networking

Accumulo is a networked application and its storage layer, HDFS, is a networked file system. Clients connect directly to tablet servers to read and write data. Tablet servers will try to read data from local disks when possible, avoiding reading data across the network, but will also often end up reading blocks of data from an HDFS DataNode over the network.

Having enough network bandwidth is fairly important to Accumulo. Even data read from disks local to a tablet server must still be transferred over the network to clients on other machines. For most clusters, servers with one or two 1-Gb Ethernet cards are sufficient. If there is more than one network interface card (NIC), they should be bonded in Linux to improve performance and availability. Currently Hadoop cannot utilize more than one NIC. Many clusters’ networks consist of a 10 Gb switch atop each rack of servers and a 10 Gb switch connecting a row of racks together.

Virtualization

Accumulo can be run on virtualized hardware, with a few caveats. HDFS makes some assumptions about the physical location of data in order to achieve good performance. If the virtual environment supports access to local disks, then these assumptions can remain valid. If, however, the physical storage of data is abstracted away onto remote media, the efforts by HDFS to reduce network I/O will be pointless. This may or may not be a problem, depending on the total I/O available.

Another consideration is server responsiveness. Accumulo continually attempts to determine the status of its processes. If server response time is highly variable due to unpredictable access to underlying physical resources, Accumulo’s timeouts may need to be increased to avoid dropping servers that are alive but don’t respond quickly enough. This increases the time that data may be unavailable before Accumulo recovers from a true server failure.

Finally, there is the issue of independence and availability. The shared-nothing architecture Accumulo is built on relies on trying to reduce dependence among hardware components in order to minimize the effect of an individual failure so that the overall system can continue functioning. In a virtual environment, a physical failure may affect more than one virtual server if those virtual servers happen to share any hardware, which can result in less availability than one might expect from separate physical machines. If these issues can be managed, Accumulo can be run successfully in a virtual environment.

Running in a Public Cloud Environment

In Amazon’s Elastic Compute Cloud (EC2) environment, for example, it is recommended that tablet server processes be run on instances with ephemeral storage, because that allows access to local disks. Some EC2 users recommend picking the largest instance type in a family, because supposedly this means that the virtual instance resources match the physical resources and that there will be only one virtual machine on a particular physical server, which can make access to the physical hardware less variable and services more responsive.

Amazon Elastic Block Store (EBS) volumes are recommended for storing the NameNode’s data, preferably across several volumes in a RAID configuration for higher availability, but not for primary HDFS storage because this increases the interdependence between servers. Alternatively, a cluster could utilize multiple NameNodes running in a high-availability (HA) configuration using local ephemeral disk.

Amazon EC2 uses Security Groups to restrict network access to specific ports and hosts. See “Network Security” for a list of ports that must be open.

Modeling Required Write Performance

For the purposes of cluster planning using back-of-the-envelope calculations, a good practice is to prototype an application and measure the performance against a single server, and then against two servers, and look for the percentage increase in read and write performance. You may have to write several gigabytes of data or more in order to test the splitting and migration properties of tables before seeing an increase in aggregate write and read rates.

Based on these rates, you can estimate the number of machines required to reach a target number of user requests to read or write application data by multiplying one or two server aggregate rates by 1.85 until the target number of requests is reached.

For example, say we needed to be able to write a million key-value pairs per second, which would allow a theoretical MapReduce job writing to Accumulo to keep up with some reporting requirements.

Testing of an application prototype on some particular hardware reveals that the single-server write rate is 120,000 key-value pairs per second. Multiplying this by 1.85, we get an estimate of 222,000 pairs per second using two servers. We continue to multiply by 1.85 until we reach 1,000,000 writes per second, doubling the number of servers in our cluster each time. At four servers we have a theoretical write rate of 760,000 key-value pairs per second. At eight servers we have a rate of 1.4 million, so we need somewhere in between four and eight servers.

A direct formula for estimating the number of servers required to reach a target write rate is as follows:

m - estimated number of machines
a - target aggregate write rate in key-value pairs per second
s - measured single server performance in key-value pairs per second

On the other hand, if you want to measure the total expected read or write rate of an existing set of servers, you can measure application performance against one or two servers and extrapolate to the size of the cluster to get the aggregate write rate.

For example, if we measure an application as being able to write 5,000 user requests per second against 1 server (where each user request translates into several key-value pairs), and we have 20 servers, we can expect to see an aggregate write rate of about 71,000 key-value pairs on 20 servers, or about 14 times the single-server write rate.

To estimate the aggregate write performance of a cluster given the number of machines and single-server performance, the following formula can be used:

a - estimated aggregate write rate in key-value pairs per second
s - measured single server write rate in key-value pairs per second
m - number of machines

Cluster Planning Example

Let’s imagine we are going to set up a Twitter clone that gets just as much traffic. Assume Twitter ingests 500 million tweets a day and each tweet is about 2,500 bytes on average. That would be about 1.25 terabytes of new data per day. Our incoming data might increase by 100 percent over the year so that by the end we’re storing 2.5 terabytes of new data per day. We need to store a year’s worth of data online and make it available for queries on this system. Data older than one year old can be deleted because it is stored on another archival system (perhaps another HDFS cluster).

So we expect to have about 685 TB of data over the course of the next year.

Using Tracing

Detailed timing information on what Accumulo is doing behind the scenes can be obtained by enabling a process called tracing. Tracing is generally only turned on temporarily while analyzing an application’s behavior, because it increases the load on Accumulo by generating data that is stored in an Accumulo trace table.

Traces can be enabled programmatically in Java or in the Accumulo shell. The Java client code is in the accumulo-core module under org.apache.accumulo.core.trace. Here is an example that uses tracing in Java:

DistributedTrace.enable(instance, zooReader, hostname, "someApplication");
Trace scanTrace = Trace.on("descriptiveString");

BatchScanner scanner = conn.createBatchScanner(...);
for (Entry<Key, Value> entry : scanner) {
  ...
}

long traceId = Trace.currentTrace().traceId();
Trace.off();

After running the trace in Java, you can either go to the Recent Traces in the monitor page or scan the trace table directory using the traceId as the row ID:

user@accumulo> scan -t trace -r traceId

Note that the trace table has the org.apache.accumulo.core.trace.TraceFormatter class configured, so the output will be more readable.

Here is an example in the shell, which returns tracing information after a pause when tracing is turned off. You can still use the monitor page or scan the trace table directly to retrieve the tracing information:

user@accumulo> trace on
user@accumulo> scan -t tablename -r rowId
user@accumulo> trace off
Waiting for trace information
Waiting for trace information

Time  Start  Service@Location       Name
 5469+0      shell@hostname shell:root
    1+5        tserver@localhost listLocalUsers
    1+2667     shell@hostname client:getTableConfiguration
    1+2674     tserver@localhost getTableConfiguration
    1+2679     tserver@localhost getTableConfiguration
    1+2686     shell@hostname client:getUserAuthorizations
   82+2723     shell@hostname scan
   41+2723       shell@hostname scan:locateTablet
    3+2743         shell@hostname client:startScan
    2+2746         tserver@localhost startScan
    2+2761         tserver@localhost startScan
    1+2761           tserver@localhost metadata tablets read ahead 4
   41+2764       shell@hostname scan:location
   39+2765         tserver@localhost startScan
   37+2766           tserver@localhost tablet read ahead 9
    1+2767             tserver@localhost open

The time column of numbers shows the length of a particular operation in milliseconds, and the start column shows how many milliseconds elapsed between the start of the entire trace and the start of a particular operation. Note that operations are nested so that the time number for one operation is greater than the sum of the times taken by the operations it initiates, i.e., those nested under it. The nesting is shown by indentation. Tracing is useful for finding operations that are taking longer than expected.

Using the Monitor

The Accumulo monitor is a convenient way to see metrics of cluster performance and diagnose problems. The monitor page shows several graphs of interest to performance analyses. Application designers can check these to see what kind of aggregate performance an application is achieving, and to verify how an application will scale when run against an increasing number of servers. For these graphs to be used effectively, estimates for the theoretical performance of individual hardware components must be known, or else any comparison to empirical performance numbers is meaningless.

In particular, looking at the aggregate ingest rate (Figure 13-1) and aggregate number of scans (Figure 13-2) will provide a good notion of how busy the cluster is.

Figure 13-1. Aggregate ingest rate in key-value pairs per second

Figure 13-2. Aggregate scan rate in key-value pairs per second

Next most useful are the graphs describing the raw amount of data written or read, in megabytes (Figure 13-3 and Figure 13-4).

Figure 13-3. Aggregate ingest rate in megabytes of data per second

Figure 13-4. Aggregate scan rate in megabytes of data per second

Monitoring CPU load can also be useful in determining how busy the cluster is (Figure 13-5).

Figure 13-5. Aggregate CPU load

Looking at caching information in the graphs at the bottom of the list is useful for seeing whether index or block caching is having the desired effect (Figure 13-6 and Figure 13-7).

Figure 13-6. Number of index block cache misses (i.e., when Accumulo has to read an index block from HDFS)

Figure 13-7. Number of data block cache misses (i.e., when Accumulo has to read a data block from HDFS)

In addition, you can measure the number of items that are being processed by Accumulo clients, which may be converting individual user requests into several key-value pairs. Knowing how to convert the number of user write or read requests into the number of key-value pairs or megabytes of data read and written will help verify that the cluster is performing optimally.

For example, if every user request to write a record of application data involves writing the original record as one key-value pair to one table, and several key-value pairs for index entries of individual fields in the record, the ratio of application records to key-value pairs written may be 1 to 10 or more.

Getting familiar with these metrics will help in reasoning about performance and tuning decisions discussed in the next few sections.

We cover the types of exceptions that Accumulo applications can expect and some ways to handle them in “Handling Errors”. When you troubleshoot an application, it is also important to be able to get information on the health of the tablet servers and master in order to determine whether the application is doing something wrong or simply experiencing issues due to hardware problems or misconfiguration.

Tablet servers forward log messages of severity WARN and above to the Accumulo monitor that displays them in one convenient location (Figure 13-8). Checking these logs can be helpful if a failure occurs within a tablet server.

Figure 13-8. Logs in monitor

When you are for hotspots, viewing the page of the monitor that shows the amount of data written or read per server is useful (Figure 13-9). If one or two servers are handling the majority of the reads and writes, versus roughly the same number of reads or writes across most of the servers, it may indicate a hotspot.

Figure 13-9. Server statistics

Further drilling down into the individual tablets hosted by a server can reveal the range of keys that reads and writes are using (Figure 13-10).

Figure 13-10. Tablet statistics

A hotspot in a particular range of keys can be a feature of some external data being loaded, over which the application has no control, or it can be the result of a deliberate key design.

In the former case, application designers can consider transforming external data into a set of keys that avoids hotspots by using techniques described in “Avoiding Hotspots”.

For more tips on troubleshooting, see “Troubleshooting”.

Using Local Logs

In addition to forwarding WARN and ERROR level messages to the monitor, tablet servers write to several logfiles on the local machine, which can include log messages of lower severity that may be useful for debugging. These are typically found in either $ACCUMULO_HOME/logs or /var/log/accumulo.

See “Logging” for details on configuring logs.

External Settings

A number of settings external to Accumulo must be configured for Accumulo to work properly. In addition to the settings described in “Kernel Tweaks” and “Software Dependencies”, the following are some settings to examine when you are tuning for performance.

 <property>
   <name>dfs.datanode.max.xcievers</name>
   <value>4096</value>
 </property>

 <property>
   <name>dfs.support.append</name>
   <value>true</value>
 </property>

 <property>
   <name>dfs.durable.sync</name>
   <value>true</value>
 </property>
 <property>
   <name>dfs.datanode.synconclose</name>
   <value>true</value>
 </property>

Memory Settings

Tablet servers are designed to serve data from disk efficiently, but they are also designed to use memory to optimize reads and writes as much as possible. Memory settings are among those that have the most significant performance impact. There are a few things to consider when configuring the amount of memory dedicated to various components of tablet servers.

[server.Accumulo] INFO : tserver.memory.maps.native.enabled = true

[tabletserver.NativeMap] ERROR: Failed to load native map library ..

Write-Ahead Log Settings

The write-ahead log currently limits the speed of writes via the BatchWriter because all mutations must be committed in an append-only fashion to the write-ahead log in order to be considered successful. Therefore, tuning the write-ahead log settings is usually worth the effort.

The following are the settings that impact performance the most:

Resource Settings

Tablet servers use system resources to respond to client requests and to perform background operations. These resources include open files, threads, and memory. We discuss memory settings specifically in “Memory Settings”.

The following is a list of settings that control the number of threads dedicated to various tasks. Consider adjusting one of these upward if the particular operation associated appears to be lagging and system resources allow. Changing the number of resources allocated to one activity may require reducing resources allocated to other activities.

Timeouts

Tablet servers keep resources around for certain periods of time to allow tasks to reuse them and avoid the overhead of setting up resources anew. These settings control how aggressively tablet servers reclaim those resources. Some of them are used to control how long to wait before determining a failure.

Timeout settings are listed in Table 13-5.

Scaling Vertically

Adding more memory and CPU to a single server will help a single tablet server process cope with more concurrent queries and writes. Modern servers can have up to 12 or more disks, which can increase the amount of CPU and RAM required to keep those disks busy.

Write-ahead logs can become a bottleneck for ingest because tablet servers each use one, albeit replicated, write-ahead log. If the ingest rate of a server is dominated by the time spent flushing mutations to the write-ahead log, adding more disks or CPU to each server will not increase the write rate. Adding more RAM and increasing the tserver.mutation.queue.max and tserver.memory.maps.max will improve performance up to a point. For this reason, it may be more cost effective to have more individual servers, each with fewer resources so that a greater portion of available disks is devoted to write-ahead logs. This is consistent with Accumulo’s design to run well on relatively cheap servers.

On individual servers that have extensive resources, including 256 GB of RAM or more and 12 or more disks, it may be possible to improve performance by running multiple tablet server processes on a single physical server. When doing so, it is important that each tablet server listen on a separate set of network ports.

Splitting Tables

Brand new tables in Accumulo start out as a single tablet. Accumulo automatically splits tablets when they reach a certain threshold known as the table.split.threshold setting in the tablet configuration. We discuss performing tablet splits programmatically in “Tablet Splits”. Here we discuss splitting tablets using the shell and additional considerations.

Some applications might be bottlenecked by the number of tablets until there are enough tablets for every tablet server to host one or more. You can choose to either wait until there is enough data ingested for the table to split automatically into the desired number of tablets, turn down the split threshold temporarily to cause automatic splits to happen sooner, or presplit the table using a set of known split points.

Lowering the split threshold temporarily has the advantage of allowing Accumulo to still pick the split points uniformly, no matter what kind of distribution of keys exists within a table. This still assumes that the splits that will occur on the initial amount of data ingested are representative of the split points that would have been chosen after all the data is ingested. For example, if we are importing a list of users that has been sorted alphabetically, the initial split points will only occur within the first few letters of the alphabet and will not be representative of how the data would be split after the entire list is imported. Users can try to obtain a representative sample of their data for the purpose of ingesting it and allowing Accumulo to find good split points early.

To lower the split threshold for a table, users can configure the table in the shell like this:

config -t tableName -s table.split.threshold=1G

Rather than lowering the split threshold, users can submit a list of split points to Accumulo to use to create multiple tablets. The advantage of this is that the table will be distributed onto more servers before any data is ingested. The onus of picking good split points rests with the user.

To presplit a table from a list of points, the split points should first be put into a text file, with one point per line. For example:

e
j
o
t

Adding these split points to a single-tablet table would result in five tablets: (-infinity, e], (e, j], (j, o], (o, t], and lastly (t, infinity) (Figure 13-11). The special tablet with end point at infinity exists for every table and is called the default tablet.

Figure 13-11. Adding split points

Then the split points file can be submitted to Accumulo for splitting via the shell:

user@accumulo> addsplits -t tableName -sf fileName

A file with split points can also be provided when first creating a table through the shell. To copy the split points from an existing table, use:

user@accumulo> createtable myPreSplitTable --copy-splits existingTable

To use a file do the following:

user@accumulo> createtable myPreSplitTable --splits-file mySplitsFile.txt

A small set of specific split points can be added directly in the shell:

user@accumulo> addsplits e j o t -t mytable

Splits can be added via the Java API as well:

 ZooKeeperInstance inst = new ZooKeeperInstance(myInstance, zkServers);
 Connector conn = inst.getConnector(principal, passwordToken);

 SortedSet<Text> splitPoints = new TreeSet<>();
 splitPoints.add(new Text("e"));
 splitPoints.add(new Text("j"));
 splitPoints.add(new Text("o"));
 splitPoints.add(new Text("t"));

 conn.tableOperations().addSplits("mytable", splitPoints);

If a set of split points are to be used to presplit a table from time to time, or to be distributed along with an application for use with multiple sources of data, care should be taken to ensure that the split points used on a table are representative of the distribution of the data to be loaded. The distribution of keys within a data set may change over time, and may not be the same from one data source to another.

Accumulo can also merge tablets, which is covered in “Merging Tablets”.

Balancing Tablets

Once a table is split into some number of tablets, the Accumulo master can use different load balancers to achieve a good distribution of tablets across tablet servers. By default, each table’s tablets are balanced separately and are assigned evenly and randomly to tablet servers. This load balancer will likely suffice for the majority of applications.

However, some applications may require more fine-tuned control over their tablets. Some key design patterns require not only that a table’s tablets are distributed evenly, but that specific subsets of tablets are also distributed evenly.

An example might be a table with a row key containing a date. Suppose an application built on top of this table frequently accesses dates falling within the same month at the same time. Using the default load balancer could end up assigning all the tablets for a month to a single server, limiting the insert and query capabilities across that group of tablets. Instead, this table could employ a custom load balancer to ensure that tablets falling within the same month are distributed evenly to tablet servers.

To write a custom load balancer, implement a class that extends TabletBalancer as described in “Additional Properties”. Add a JAR containing the new balancer to Accumulo’s CLASSPATH on at least the master nodes, and configure a table to use this balancer by setting the table.balancer property for the table.

Balancing Reads and Writes

Accumulo allows users to dedicate system resources to writes or reads as necessary. By default, Accumulo does not throttle writes in order to keep some resources available for reads, so query performance can suffer if clients are using too much of Accumulo’s resources for writing.

One method of throttling writes and improving read performance is to decrease the maximum number of files Accumulo is allowed to create per tablet. This is controlled with the per-table setting table.file.max, which defaults to 15. When the maximum number of files has been reached for a tablet, Accumulo will merge new data with data from one of the existing files instead of creating a new file for that tablet. The new, merged file will then replace the existing file. This process is called a merging minor compaction.

When merging minor compactions are occurring, the overall write rate of the cluster may begin to decrease, because a minor compaction to free up memory for new writes may be waiting for a merging minor compaction to complete. While this is happening, any new writes to the server cannot proceed and clients are told to wait. Increasing the table.file.max setting or decreasing the table.compaction.major.ratio setting will ensure that enough background compactions occur so that minor compactions will not end up having to wait and block clients.

Data Locality

In MapReduce jobs, the notion of physical data locality, in terms of the distance between data to be processed and the CPU and RAM elements in which it will be processed, is extremely important. Many types of MapReduce jobs are run on data that is so large that reading it from some storage medium over a network would limit the performance in an unacceptable way. The primary innovation of MapReduce versus many other types of data processing is that it sends the computation to the data, rather than moving the data to the computation.

When key-value pairs are processed in a MapReduce job, the key-value pairs are read from a local disk by a copy of the map or reduce process that has been sent to the machine holding the data. This allows the overall job to be limited by the aggregate throughput of all the hard drives in the cluster, which is often much higher than the aggregate throughput rate of the network connecting machines in the cluster.

Data for MapReduce jobs is stored beforehand in HDFS, which automatically distributes it over many machines, handles replication, and helps programs find a particular piece of data by exposing the IP address of the physical machine on which it is stored.

For Accumulo, the concept of physical data locality is still important but not paramount. Because Accumulo uses HDFS to store files, each time a tablet server flushes a new file to HDFS as part of the minor compaction process, one copy of that file is stored locally on the machine hosting the tablet server process. Subsequent reads can then simply read from local disk rather than pulling data from a remote machine over a network.

Over time, as the Accumulo master performs load balancing of tablets, some tablets may reference files for which there is no local copy, forcing reads to pull data from a remote machine over the network. But eventually the major compaction process will tend to create new files for each tablet, merging several old files into one new file, which will cause a local copy to be created. Good physical data locality is something Accumulo achieves eventually and asynchronously.

Accumulo has a utility for checking the level of data locality in a cluster. It can be run via the accumulo command:

accumulo org.apache.accumulo.server.util.LocalityCheck -u root -p secret

Server         %local  total blocks
 10.10.100.1 100.0        7
 10.10.100.2 100.0       10
 10.10.100.3 100.0       10

If for some reason a cluster has poor data locality, increasing the frequency of major compactions or scheduling a major compaction can cause files to be rewritten. To compact a table, use the compact command from the shell:

accumulo@cluster> compact myTable

Note that this will completely rewrite all files in a table. When stopping and restarting a cluster, Accumulo tries to reassign tablets to the same tablet servers that were previously hosting them before shutting down, so that physical locality is preserved.

Sharing ZooKeeper

Some other applications running on or close to the Accumulo cluster might also make use of ZooKeeper. In particular, Apache Kafka is a popular distributed queue used to stream data to and from applications. Some Accumulo clients can make use of Kafka to stream data to Accumulo or from Accumulo to other applications.

Similarly, Apache Storm, a popular streaming data processing framework, relies on ZooKeeper for configuration information.

Be cognizant of the load placed on ZooKeeper by these other systems. (See the ZooKeeper documentation on monitoring.) Although ZooKeeper itself is a distributed application, one machine of the quorum serves as master, meaning that all writes go to it, whereas reads can go to any member of the quorum. The ZooKeeper quorum master has to handle all writes and replicate those writes synchronously to other members of the quorum, so the scalability of writes isn’t improved by adding more machines; rather, it makes each write more expensive, reducing the total write throughput of the ZooKeeper instance.

If ZooKeeper can’t keep up with operations, Accumulo will not function properly. It is possible to run multiple separate ZooKeeper instances on a cluster, each one consisting of one, three, or five nodes, as long as applications are configured to use different instances.