Showing posts with label Cassandra. Show all posts
Showing posts with label Cassandra. Show all posts

Wednesday, October 11, 2017

HBase, Cassandra and CAP


Which is best - Cassandra or HBase? Well, it depends what you're trying to do.

The oft-quoted comparisons generally involve Eric Brewer's CAP-theorem (the formal proof by Gilbert and Lynch can be found here and an updated review here) and you can see pictures like this dotted around the web:

From "Cassandra: the Definitive Guide" quoted here
(where the RDMBS offerings imply Two-Phased Commit).

Most developers have some awareness of CAP but there are a lot of misconceptions - typically that you can have at most two elements of Consistency, Availability or Partition tolerance - but not all three. "The 2 of 3 formulation was always misleading because it tended to oversimplify the tensions among properties" (from an article by Brewer himself found here).
Allowing at least one node to update state will cause the nodes to become inconsistent, thus forfeiting C. Likewise, if the choice is to preserve consistency, one side of the partition must act as if it is unavailable, thus forfeiting A. Only when nodes communicate is it possible to preserve both consistency and availability, thereby forfeiting P. The general belief is that for wide-area systems, designers cannot forfeit P and therefore have a difficult choice between C and A.
Here are some common misconceptions corrected.

  • The C in CAP "has got nothing to do with the C in ACID, even though that C also stands for 'consistency'" [1]. "Gilbert and Lynch use the word “atomic” instead of consistent in their proof, which makes more sense" [2]. The C in ACID has several similar but not mutually exclusive meanings which include not violating any database constraints, linearizability (see below), ensuring the data is consistent (eg, domain-specific rules such as taking X out of one bank account and into another means the total money overall does not change).
  • "The CAP model says nothing about transactions that touch multiple objects. They are simply out of scope for the theorem" [2].

In "Please stop calling databases CP or AP", Martin Kleppmann says:

"But if you’re using some other notion of consistency or availability, you can’t expect the CAP theorem to still apply. Of course, that doesn’t mean you can suddenly do impossible things, just by redefining some words! It just means that you can’t turn to the CAP theorem for guidance, and you cannot use the CAP theorem to justify your point of view.

"If the CAP theorem doesn’t apply, that means you have to think through the trade-offs yourself. You can reason about consistency and availability using your own definitions of those words, and you’re welcome to prove your own theorem. But please don’t call it CAP theorem, because that name is already taken."

In the CAP theorem:
  • Consistent means Linearizability,  think of visibility in terms of variables in the JVM. "If operation B started after operation A successfully completed, then operation B must see the the system in the same state as it was on completion of operation A, or a newer state." [1] A master that asynchronously replicates to slaves is an example of a non-linearizable system. Note: some systems (eg a DB that uses MVCC) are intentionally non-linearizable.

  • Availability means “every request received by a non-failing [database] node in the system must result in a [non-error] response” [from the proof]. Note, the response can take an arbitrary amount of time which might violate some people's notion of availability...

  • "Partition Tolerance (terribly mis-named) basically means that you’re communicating over an asynchronous network that may delay or drop messages." Which is basically the internet. "

    "In practical terms, a distributed system cannot be made immune to network failures... If the system chooses to favor Availability (such systems are designated AP systems), then it will continue to service requests, even though the data could be in an inconsistent state... If the system chooses to favor Consistency (known as a CP system), then it will choose to stop serving requests (become unavailable) if the consistency of data cannot be guaranteed. For a CP system, this is achieved by requiring a certain number of nodes to confirm that a data update has been made before acknowledging the update." [Hazelcast]
Using these definitions, Cassandra is not AP for quorum read/writes. If there is a partition split, the nodes on the wrong side of the split cannot reach a consensus and therefore are not available. (Interestingly, Brewer notes "as some researchers correctly point out, exactly what it means to forfeit P is unclear.")

So, although Cassandra and HBase both have their pros and cons, mentioning the CAP theory is orthogonal.

(Aside: although Elastic Search is a search engine not a database, there is an interesting chat about how CAP applies to it here.)

A solution?

There's a cheekily entitled blog post (it's titled "How to beat the CAP theorem" but in it he says "You can't avoid the CAP theorem, but you can isolate its complexity and prevent it from sabotaging your ability to reason about your systems"). This isolation is via immutability.

In this post, author Nathan Marz talks of DBs "choosing availability over consistency":
The best consistency guarantee these systems can provide is eventual consistency. If you use an eventually consistent database, then sometimes you'll read a different result than you just wrote. Sometimes multiple readers reading the same key at the same time will get different results... It is up to you to repair the value once you detect that the values have diverged. This requires tracing back the history using vector clocks and merging the updates together (called read repair)
Marz' proposal to use immutable data leads us to this conclusion:
If you choose consistency over availability, then not much changes from before. Sometimes you won't be able to read or write data because you traded off availability... Things get much more interesting when you choose availability over consistency. In this case, the system is eventually consistent without any of the complexities of eventual consistency. Since the system is highly available, you can always write new data and compute queries. In failure scenarios, queries will return results that don't incorporate previously written data. Eventually that data will be consistent and queries will incorporate that data into their computations.
This mitigates Brewer's proposal that revolves around reconciliation during a partition recovery phase. Marz' proposes no reconciliation, just a deferred sharing of the data.

[1] Martin Kleppmann's blog.
[2] Julian Browne's blog.

Friday, April 17, 2015

Cassandra Table Design


Our customer wants a table that has hundreds of thousands of rows stored in it per day. No problem, Cassandra handles it well. Our performance test shows that with a Replication Factor of 3, writes in a 5-node cluster take a touch less than 1ms and reads about 2ms per row. If you want to pull out more rows, then it's even better - 1000 rows takes about 400ms.

However, our customer wants to slice and dice the data, pivoting on an arbitrary set of columns, aggregating values as they go. Now, Cassandra is great at storing data but does not offer the richness you'd get in a SQL DB for retrieving subsets of it. You really want to search on primary keys. You can create indexes but they just create invisible tables for you. And even if you search on primary keys, you need the right permutation of keys to avoid WITH FILTERING.

So, our first stab got us creating way too many tables because of our customer requirements. The number of column families we can pivot on causes the number of tables to grow factorially. If our customer requires us to pivot on say 5 columns, that's a combination of

n!
r!(n-r)!

where n in our case is the 5 potential choices on offer and r is the number the user chooses.

Of course, our user can choose any number, from 1 to 5, so the number of permutations now looks like:

 5 

r = 1
n!
r!(n-r)!


Which is starting to cause the number of tables to balloon since

up to 3 choices means 7 tables
up to 4 choices means 15 tables
up to 5 choices means 31 tables
.
.

The number of rows also starts rising like:


 
r = 1
 r 

p = 1

dp


where r and p are just indexes and dp is the number of distinct elements for a given dimension p.

Note: this is the worse case scenario. It depends on your data how many rows there will actually be. As it happens, since our data "clumps" around certain regions, our numbers were much smaller than this.

But this article has a very interesting way of querying the data for just what you want and using one table in which to do it. To leverage it, we need to pre-compute the aggregations. As it happens, we're using Apache Spark for that then dumping it into Cassandra.

Thursday, February 26, 2015

Cassandra for the Busy - Part 2


Interesting problem I had this week that's obvious when you think about it.

The Config

I have a cluster of 5 Cassandra (version 2.0.11.83) nodes. I execute:

DROP KEYSPACE IF EXISTS my_keyspace
CREATE KEYSPACE my_keyspace WITH REPLICATION = { 
    'class' :              'SimpleStrategy',
    'replication_factor' : 2 
}
CREATE TABLE my_table (
    my_int_id    varint,
    my_string_id varchar,
    my_value     decimal,
    PRIMARY KEY (my_int_id, my_string_id)
)

I then start hammering my cluster with some CQL like:

INSERT INTO my_table 
    (my_int_id, my_string_id, my_value)
    VALUES (1, "2", 42.0)

or such like. The code I use to write to the DB looks something like:

RegularStatement  toPrepare         = (RegularStatement) (new SimpleStatement(cql).setConsistencyLevel(ConsistencyLevel.TWO));
PreparedStatement preparedStatement = cassandraSession.prepare(toPrepare);
.
.
BoundStatement    bounded           = preparedStatement.bind(new BigInteger(myIntId), myStringId, new BigDecimal("42.0"));
cassandraSession.execute(bounded);


The Error

I then kill a single node and my client starts barfing with:

Not enough replica available for query at consistency TWO (2 required but only 1 alive)

which baffled me initially because there were still 4 nodes in the cluster:

dse$ bin/nodetool -h 192.168.1.15 status my_keyspace
Datacenter: Analytics
=====================
Status=Up/Down
|/ State=Normal/Leaving/Joining/Moving
--  Address       Load       Tokens  Owns   Host ID                               Rack
UN  192.168.1.15  83.98 KB   256     40.0%  86df3d65-0e19-49b2-a49e-bc89d6347fa4  rack1
UN  192.168.1.16  82.78 KB   256     42.0%  8fghe485-0e19-49b2-a49e-bc89d6347fa5  rack1
UN  192.168.1.17  84.91 KB   256     38.0%  9834edf1-0e19-49b2-a49e-bc89d6347fa6  rack1
UN  192.168.1.18  88.22 KB   256     41.0%  873fdeab-0e19-49b2-a49e-bc89d6347fa7  rack1
DN  192.168.1.19  86.36 KB   256     39.0%  2bdf9211-0e19-49b2-a49e-bc89d6347fa8  rack1

Of course, the problem is that your tolerance to a node fault is:

    Tolerance = Replication Factor - Consistency Level

For me, that was Tolerance = 2 - 2 = 0 tolerance. Oops.

This is not a problem on all inserts. Running

dse$ bin/cqlsh -uUSER - pPASSWORD
Connected to PhillsCassandraCluster at 192.168.1.15:9160.
[cqlsh 4.1.1 | Cassandra 2.0.11.83 | DSE 4.6.0 | CQL spec 3.1.1 | Thrift protocol 19.39.0]
Use HELP for help.

and inserting worked fine. But this is because cqlsh uses a different consistency level.

cqlsh> consistency
Current consistency level is ONE.

so, let's change that:

cqlsh> consistency TWO

Consistency level set to TWO.

Now do the inserts fail? Well sometimes. It depends on which nodes we attempt to write to, something transparent to the user.


The Problem

Cassandra will try to put a piece of data on REPLICATION_FACTOR number of nodes. These nodes will be determined by hashing their primary key. If a node is down, Cassandra will not choose another. It simply cannot be written to at this time (of course, it might get the data if/when the node joins the cluster).

Upon inserting, the client decides how many nodes must be successfully written to before the INSERT is declared a success. This number is given by your CONSISTENCY_LEVEL. If the contract cannot be fulfilled then Cassandra will barf.


Further Reading

Application Failure Scenarios with Cassandra

Lightweight Transactions in Cassandra 2.0


Monday, February 23, 2015

Cassandra For the Busy - Part 1


A quick guide to Cassandra admin

From the Cassandra installation directory, running:

bin/nodetool -uUSER -pPASSWORD -hNODE ring KEYSPACE

(where the username, password, node and keyspace correspond to whatever you set up) produces data that reads like this:

Token is a number between Long.MIN_VALUE and Long.MAX_VALUE that is used in hashing the key. (See org.apache.cassandra.dht.Murmur3Partitioner.describeOwnership for how ownership is calculated over  a cluster using something like a MurmerHash algorithm. This is the default for Cassandra ).

Load refers to how much has been flushed to disk. Therefore, this may take some time to register a change after you have truncated a table containing 10 million rows. This appears to correspond to org.apache.cassandra.service.StorageService.getLoad().

Owns This represents how much of the ring range a token covers. Note: it might not add up to 100%. "For DC [data centre] unaware replication strategies, ownership without replication will be 100%" (see code comments here. for org.apache.cassandra.service.StorageService.effectiveOwnership). But with replication, it appears to be 100% * the replication factor.