It's time to take another look at MongoDb

MongoDb is one of the most popular database systems today and certainly the most widely known document store solution.  Since the last version (3.0) has been released this month, I've decided to write a post on what are its strengths and weaknesses (in my opinion) and some tips in writing Mongo client applications.

The bad news

 

Despite the hype, a strong wave of criticism has emerged in the last years pointing out several important flaws:

Unlike most database systems, durability is configurable in MongoDb, and more importantly, the default value does not assure it. For durability I'm referring to the basic guarantee that, if a write to the database is successful, the data has been stored and if it if was not due some problem the client is dully notified (we are not talking here about replica based fail-overs during a crash or eventual consistency between replicas). The universal mechanism used for that is a write-ahead-log that keep tracks of the operations pending to be processed, a write is only successful if it has been recorded in the log. In MongoDb this mechanism is called journal and by default clients does not wait for the journal to be written to return from a write operation. The operation stays in the server memory till a thread dumps it to the journal asynchronously, thus we have a time window where , if the server crashes, there is an undetected data loss. Fortunately you can configure the client to wait to to the journal write before returning a success, but it is not the default behavior, and it takes a performance hit. Most of the initial MongoDb benchmarking were done without journaling making the comparison to other systems a bit unfair.

Initially MongoDb write locking system was global meaning that the whole mongod server instance was blocked for each write effectively serializing all the insertions, in later releases the lock changed to database level, forcing in most cases to model the data using a unique collection per database to parallelize data insertion. 

Scalability. Mongo is not the ideal repository when thinking in web-scale requirements, certainly not when compared with systems designed to store distributed data sets on the petabyte range like Hbase or Cassandra. Despite of the marketing and branding name (Humongous) it does not seem that it was initially designed with cluster distribution in mind: when using sharding and replication, you need different replica node sets for each shard: for example, a total of 9 different nodes is needed for supporting 3 shards with a replica factor of 2. the configuration is also complicated needing separate managing processes (mongos) also configured redundantly in HA, and other extra processes to store and serve the cluster metadata (configuration manager) deployed in a cluster of 3 nodes. On top of that the database functionality is limited when using sharding. 

Storage. MongoDb is schema-free but not schema-less, storing the documents in BSON format means that each document is stored with its own schema: field names, structures, hierarchy relations.. This is a great storage overhead compared to other schema bound solutions.

The good news 

 

Obviously a product so successful had several strong points:

The document format, BSON can be directly mapped from JSON, this is a huge advantage for front end development based on javascript: the ajax messages can be directly stored in the repository. There is no relational mapping , no JPA needed.

Schema flexibility means that there is no problem with schema changes, new fields, field removal...,  One of the most painful scenarios we can found when using a relational database is gone.

This is in part possible because there are no relation between collections, no multi-collection transactions or joins.

New developments

This month MongoDb published the new 3.0 release, including a new storage engine, WiredTiger, somewhat fixing two of the most criticized points:

WiredTiger supports compression (two codecs, snappy and zlib). This is a huge storage and disk IO improvement. Mongo claims that data storage can be reduced in a 70%. In our particular case I tested the size reduction in more than an 80% using Snappy. (Snappy aims for fast compression and reasonable compression whereas zlib provides maximum compression but is slower). The tests also showed improvements in the average writing speed the reduced disk I/O compensated the cpu time used in compression.  My case is specially favorable since the documents are very big and have a high degree of redundancy.

WiredTiger write locking is done at document level providing the highest throughput. Even if you use the former mmap engine, locking is now done at collection level.

Building our stack

 

MongoDb is a good solution when you have to store unstructured data. It also gives you the possibility of horizontal scaling and high availability out of the box. Compression was a critical improvement since data redundancy is inherent to the model: there are no relations, nor joins, so all needed data has to be stored in each collection even if duplicated in a de-normalized fashion; the schema is free, meaning it is stored with each document. The reduced storage needs imply that sharding can be minimized or disregarded and an increment of the scale out limits.

There is no excuse for not using Object Document Mapping

When using MongoDb from Java the first thing that we notice is the impedance mismatch between the java object and the BSON document. This is can be addressed using an object document mapper. Unlike ORM where the impedance with the relational model can be very high, and there are advocates of either using directly SQL or a mapping framework like JPA, there is no discussion here that the best approach is to use a document mapper, since each document will be stored in a collection, not normalized, there will be no relations, external keys, more than one table involved...

Use Jackson

Usually the only variable measured when choosing a json mapper is the performance. It is a main factor for sure, but the feature set provided is equally important. Jackson is in the top regarding the serialization speed and is, hands down,  the more flexible and cofigurable, providing the richest feature set. With Jackson you even don't need to annotate classes directly or you can do it externally using  Mix-in.

Leverage the existing frameworks

There are two frameworks that provide databinding using jackson as the json mapper: mongojack and jongo. The two of them serialize directly to BSON removing an intermediate step. I opted for mongojack since the approach is to map a collection to a generic type in creation, and jongo needs to receive the destination object class on each query call, this can be useful if you need to change it for different queries. Either once in creation or in each query call the class must be provided due to Java type erasure.

Keep it simple, your data is never so unstructured

Use the class definition as the schema of the collection, (that's the approach with mongojack), mapping one type to one collection, leverage inheritance so instances of extending classes can be stored in that collection representing evolving schemas. Jackson supports the storing of the class type as metadata in the document to provide polymorphic deserialization, not existing fields can be configured to be ignored, I have tested it with mongojack and it works perfectly.

Use ObjectId as the key for all the collections, look at it as a sequence or auto-incremental synthetic id guaranteed to be unique in the cluster and containing the document creation timestamp.

Write in batches

Although MongoDb provides a configurable durability model, the reality is that the vast majority of applications will need some storage guarantees, that is a journaled write concern.

This is implemented at MongoDb server side as an scheduled thread that writes to the journal the operations queued in memory periodically. If we write in batches, aside of  the benefit of minimizing network trips, we maximize the number of operations that will be saved to the journal in the next period.

Your mileage may vary

I arrived to this guidelines after analyzing the different storage requirements of several components in our architecture and examining MongoDb capabilities and limitations. I designed a general data abstraction layer using this ideas targeting both ease of use and performance.

While I think that this ideas can be applied to most of the cases, your scenario may need a different approach (or a different database solution). In any case I think that, even if you discarded Mongo before, with the improvements added to the last release, may be it is worth a reassessment.

Distributed Storage Concepts in Vertica and Cassandra

In this post I want to explore the topic of distributed storage by comparing two different products: Apache Cassandra, an open source operational database system and Hp Vertica a proprietary analytic database system.

A distributed database is a computer network where information is distributed and stored in several nodes.

Cassandra is a column-oriented distributed storage system with no single point of failure capable of scaling out to hundred an even thousand of nodes, highly available and resilient to network splits. It is  designed to support a high ratio of random writes without sacrificing the read efficiency providing eventual consistency.

I've had previous experience working with Cassandra but recently I became involved  in a project where we needed a distributed storage solution to store time series and perform analytic queries on the data. While comparing different products we came across Hp Vertica.

Vertica is the commercial version of the C-Store columnar storage system, an academic collaboration between different universities.

What sets Vertica apart from the new batch of NoSql distributed storage solutions, is that it is actually an RDBMS, ACID and fully SQL compliant and capable of scale out to large clusters. There are deployed production clusters in the range of hundreds of nodes and over the petabyte size. Vertica is an analytic database, and as such is designed to ingest great amounts of data in batch processes and support a relatively low number of very complex read transactions.

The two products are share-nothing, elastic, scale-out architectures.

The Consistent Hash Ring

When we think about a distributed storage solution, our first step is to define the sharding strategy we are going to use to split the data among nodes of the cluster.

The simpler solution could be to arbitrarily map ranges of keys to nodes. This is actually a working approach used in some systems but it has disadvantages, namely the need to store and maintain metadata tables (one per objetct/table/collection key type) with the mapping of ranges to nodes. It would probably need a central point of coordination.

We can also simply apply the function -key value- modulo -number of nodes- and the result would be the node containing the key.  Since we need an integer value and also to avoid hotspots caused by the skew of our key distribution first we will need to apply a hash function with a good mixing behavior to the key: hash(key value) modulo -number of nodes-. A good candidate can be MD5 hash. Now we have a good performing algorithm that distributes the data evenly in our cluster.

What happens if we add a new requirement?. We need our cluster to be elastic, nodes can be added or removed as needed and the data must be re-balanced in such cases. With our current strategy when we add or remove nodes, we will need to re-hash an move a huge amount of data. We need a strategy decoupled from the number of nodes.

We already introduced the hash function to make or solution independent of the key types and skew. Now we introduce the concept of consistent hashing: we to apply the same hash function we use for the key value to the node id (the id can be an unique property like the ip address) . We map each node as a value in the hash range, since this range, although huge, is limited we can handle it as if it wraps up in a circular way forming a ring: when we reach the maximum value the next one is 0.  We have our nodes represented as points inside the ring. To find the node corresponding to a key, we apply the hash function to the key so we get also the point representation of the key in the ring, then we move clockwise until we find the following node point and that's the one storing the data corresponding to this key. The consistency means that when we add or remove a node we only need to relocate the data stored on it.

This solution is considered the canonical consistent hash ring strategy. The node is randomly assigned a point in the ring by applying the hash function to the node id, the risk of collision with another node is negligible, the trade off is that we don't know what will be the range assigned to a node, and there could be great size differences, creating hot spots. This is particularly patent with a small number of nodes. Ranges tend to even up with a large number of nodes due to the good mixing behavior of the hash function. 

The problem of load balancing in the cluster can be resolved using the virtual nodes strategy. We split the hash range or continuum in a big fixed number of slices of the same size called virtual nodes. The keys are hashed and assigned to a virtual node by applying the modulo function. The same number vnodes are assigned to the physical nodes thus avoiding hot spots. We can even assign more vnodes to the most powerful server if out cluster is not homogeneous. The mapping between nodes an vnodes is maintained in a metadata table. This is an hybrid of the two first proposals. The distribution depends on the number of vnodes, but it is a fixed value during all the life of the cluster. The mapping metadata table to maintain is unique.

Cassandra Partitioner

Cassandra initially used a partitioner to decide the node the data is stored on. (In fact the system even allows to partition the data arbirtrarily following the key ordering sequence although is strongly not recommended.

From the documentation:

"A partitioner determines how data is distributed across the nodes in the cluster (including replicas). Basically, a partitioner is a hash function for computing the token (it's hash) of a row key. Each row of data is uniquely identified by a row key and distributed across the cluster by the value of the token"

At first Cassandra followed the consistent hash ring with pseudo random generated token ids strategy. As we saw previously this solution leads to a load balancing problem in the cluster. Beginning with version 1.2 Cassandra introduces vnodes:

"Prior to version 1.2, you had to calculate and assign a single token to each node in a cluster. Each token determined the node's position in the ring and its portion of data according to its hash value. Starting in version 1.2, Cassandra allows many tokens per node. The new paradigm is called virtual nodes (vnodes). Vnodes allow each node to own a large number of small partition ranges distributed throughout the cluster. Vnodes also use consistent hashing to distribute data but using them doesn't require token generation and assignmen"

In the latest releases the virtual node strategy was adopted.

Vertica Segmentation

Vertica reserve the use of the partitioning concept to local intra-node tuple segregation to distinguish from inter-node segregation.

From the original Vertica architecture paper:

"Vertica applies a default hash function to the columns chosen as segmentation keys, This function distributes the data using a normal statistical distribution. The node on which the tuple is stored is determined by this hash. The whole range of hash values is divided between the number of nodes, and each node is assigned a range beginning with the previous node maximum and covering the maxInt/numNodes (max Integer value is 2^64) following values."

This is the second approach we described, this is dependent on the number of nodes, and any node joining or leaving the cluster force us to relocate almost all the data.

Lately Vertica introduced a new feature called elastic cluster:

"To help make data re balancing due to cluster scaling more efficient, HP Vertica locally segments data storage on each node so it can be easily moved to other nodes in the cluster. When a new node is added to the cluster, existing nodes in the cluster give up some of their data segments to populate the new node and exchange segments to keep the number of nodes that any one node depends upon to a minimum"

"The alternative to elastic cluster is to re-segment all of the data in the projection and redistribute it to all of the nodes in the database evenly any time a node is added or removed. This method requires more processing and more disk space, since it requires all of the data in all projections to essentially be dumped and reloaded."

These local segments correspond to the virtual nodes strategy.

Conclusion

It seems that we have a winner:The virtual nodes solution. Vertica and Cassandra arrived to the same strategy to distribute the data in the cluster. In fact, this solution is widely used in similar systems like Voldemort or DynamoDb.

Revisiting the CAP theorem

You are probably familiar with the CAP theorem, in brief it states that a distributed system cannot simultaneously guarantee the following properties instead it must pick two of the three and neglect the third:

• Consistency. All nodes see the same data at the same time.
• Availability. Every request receive a response either success or fail.
• Partition. The system is resilient to network partitions and continues operating in such cases (split brain situation).

According to this, systems can be classified as CA, CP or AP.

The first time I heard of it I remember thinking that A was not possible in a real distributed systems (see the fallacies of distributed computing) and that A and P seemed to somewhat overlap.

Later I came across this interesting post by Coda Hale that re-explains the theorem in a more sensible way:

Distributed Systems are defined by the property they choose to guarantee when there is a network partition: Consistency or Availabilty. Systems can be:

CP: The system chooses consistency over availability on a network partition, I the event of a network split the system stops working or return error to all requests.

AP: The system chooses availabilty over consistency on a network partition, that means that nodes can still being giving service indepently. Once the connection is recovered, a synchronizaton mechanism restores a consistent view. Take in account an AP system is giving up strong consistency in favor of soft or eventual consistency for all its operational life to avoid a service outage in the split network exceptional situation.

The corollary is that a distributed system cannot be CA and guarante both consistency and availability in the event of a partition. Only a not-distributed system (a unique node) can be CA.

Coda Hale reasons also that availabilty is preferable over consistency in most systems since a service outage has always an economic cost and strong consistency is rarely required.

On the other hand, Michael Stonebraker one of the creators of Vertica, favors consistency over availability, his argument is also solid: the split brain situations are very rare, and in favoring availabilty over consistency in such cases you are sacrificing consistency also in nomal operational situations that are the vast majority of cases.

Cassandra is an AP system designed to be highly available, and that is a priority in an operational database system. Although it supports a tunable consistency model trading off latency for consistency, it is intended to work as an eventual consistency system and does not support row locking in any case.

It is no surprise that Vertica is a CP system, it provides an ACID consistency model. and since it is an analytical db where availability is not the priority, it seems like a right choice.

Storage Considerations

A great deal of the differences of behavior and performance come from the different local storage policies chosen, I'll leave that for a future post.