NoSQL Databases

4.1 Why We Move Away from RDBMS

4.1.1 The ACID promise and the system of record

Before NoSQL, the storage that was available was largely the RDBMS — the relational database management system. These systems were typically meant for OLTP — online transaction processing — and were also known as the system of record, because they carried out day-to-day business transactions. The main concern with an RDBMS is strict consistency and durability guarantees over the multiple data items involved in a transaction. To deliver those guarantees, an RDBMS commits to the ACID properties, which were discussed in detail in earlier sessions:

• A for atomicity — a transaction either happens fully or not at all.
• C for consistency — a transaction moves the database from one valid state to another.
• I for isolation — concurrent transactions do not interfere with each other.
• D for durability — once a transaction is committed, it stays committed.

Sticking to these strong guarantees is also what makes an RDBMS hard to scale.

4.1.2 Vertical scaling and the volume bottleneck

Because RDBMS systems strictly follow ACID, they face trouble in scaling. They can scale to some extent, but only at a very high cost, because they follow a kind of vertical scaling: if you want to increase storage capacity you have to stop your system, add newer, high-end hardware — good quality hard drives, maybe SSDs, or something better — and bring the system back up. They do not follow the horizontal scaling that distributed systems follow, which is very easy to scale: you just add another machine. Vertical scaling is expensive, disruptive, and does not grow cheaply.

The RDBMS is also not suited for applications with very large volumes of data, because at some point the performance bottlenecks when the system has to deal with a very large volume of data. And if you want to build distributed, geoscale applications — applications hosted across the globe — the RDBMS is not the right fit either.

4.1.3 Applications that can relax consistency

There are scenarios that simply do not need strict consistency and durability for every use case. Take a social media application running on multiple data centers across the world. The application is hosted in several geographies, and some application data stays in one region while other data stays in another region, with a sync happening at some point in time — not at the same moment the write happened. Consistency is going to be compromised in certain ways, and that is acceptable.

Real-world: You have probably hit these compromise points yourself. While shopping on an e-commerce site and trying to add an item to your cart, sometimes the add does not work — a rare event, but it happens. If you hit a like button on a post, it is possible the like is not recorded. There are multiple reasons for these failures. The point is that these systems do not require much consistency; the focus is not on strict consistency.

The same idea applies to an e-commerce website: a credit card transaction absolutely needs strict consistency, but browsing products or adding items to a cart does not. Those are the scenarios where other technologies — apart from the RDBMS — can simulate the behavior with much better performance. That is why we move to NoSQL: better performance for the cases that can relax consistency.

4.1.4 Schema rigidity: the uniformity assumption

The second reason is schema. In an RDBMS, every table has a schema, and the collection of tables forms the database. One table maintains uniformity across its records: if a table has \(n\) attributes \(A_1, A_2, \ldots, A_n\), every record inserted into that table must carry all \(n\) values. This uniformity is a guarantee, but there are scenarios where this much uniformity is not required — each row may carry a different kind of data that you want to store.

There is also the connectivity problem: certain types of information involve extreme connectivity. In a social media application, how many friends do you have between two hops? In a network topology, routers are connected to switches and to one another — a huge, complex topology. For scenarios like these, again, the RDBMS is not a good choice. For all such scenarios there is a fourth category of database — very, very important — the graph database.

4.1.5 Where RDBMS is still the right fit

The RDBMS is the right fit when you want normalized tables and you need to perform many join operations. That is its home turf. But when you want to store flat structures, do not want frequent joins, and do not want strict consistency levels, you may go for one of the other technologies — and based on the scenario given to you, you choose one of the NoSQL databases to build your application. The choice is scenario-driven: no single storage engine is right for everything.

4.2 What Is NoSQL: Origins and Shared Characteristics

4.2.1 The name: 1998 and 2009

The term NoSQL has two birthdays. It was first given by Carlo Strozzi in 1998 — a pretty old term — to describe a lightweight open-source database without a standard SQL interface. Then in 2009 the term was reintroduced, with the meaning we use today: non-relational databases that do not follow the relational ACID properties or the rigid schema thing — they follow something different. The name is best read as Not Only SQL: it is not the absence of SQL, it is SQL plus other things.

4.2.2 The characteristics shared by NoSQL systems

Around those two points in time, many big tech companies were working on their own storage technologies, and each came up with its own design. Yet the technologies that gathered under the NoSQL name shared a set of common properties:

• They were not only SQL based.
• They were non-relational in nature — the concept of foreign keys linking two tables was not given priority.
• They were schema-less — no rigid schemas.
• They were trying to loosen consistency to address the scalability and availability requirements of large-scale applications.

What does "loosening consistency" look like in practice? A NoSQL system often runs in a distributed setup. If somebody hits a like on the nearest server and that server dies, the like is not communicated across the multiple machines, and the write may not succeed. If you are adding an item to a cart and that happens to be the last item, another user may have added it meanwhile, so your add fails. Any of these things may happen — and that is what we mean by relaxing consistency.

4.2.3 The performance motivation behind the movement

The NoSQL databases were born out of the open-source movement of large web-scale applications. Many people wanted to build large-scale web applications that cater to huge user bases. The builders looked deeply into the databases and found where the performance bottleneck was: the RDBMS, with its strict consistency requirements, was the bottleneck. The question was — if I change the storage, will my performance improve in certain scenarios?

The answer is careful, not blanket. Not every scenario can compromise consistency, and the RDBMS is not obsolete. There are real-life scenarios that do not require consistency and durability, and for those we move to other technologies for better performance — technologies that compromise on consistency and durability. In place of strict consistency, they can go for eventual consistency: the system converges to a consistent state later, after the writes settle.

4.2.4 Throughput, distribution, and cluster friendliness

Two more characteristics matter. First, these databases are distributed for scale: they can be hosted on multiple nodes, and you add a node to scale the system easily — horizontal scaling. Second, they are cluster friendly: they can work as one form of a cluster that coordinates to achieve a task. These characteristics were observed across all the databases we will talk about. Yahoo came out with technologies, Google came out with technologies — and a conference was held to look into all of them, which is where the name NoSQL (Not Only SQL) was settled for the group.

4.2.5 A motivating example: flash sales

Real-world: A few years back, when OnePlus phone flash sales used to happen, the phone went on sale for one or two seconds and the websites used to crash — the system could not cater to that much data arriving in a burst at a certain moment. The companies later improved their applications, and such events are rare now. The lesson is the motivation for NoSQL: we want to process a large volume of data while keeping performance in mind — high throughput without bottlenecking.

4.3 The NoSQL Data Model: Variety Without Schema

4.3.1 Structured, semi-structured, and unstructured data

What is the data model for NoSQL databases? NoSQL databases support storing a variety of data. Variety here typically means three different kinds:

• Structured data — the kind the RDBMS was typically used to store, with a rigid, pre-defined layout.
• Semi-structured data — data that has some structure (for example, tags or fields) but not a rigid table layout.
• Unstructured data — text, images, and other content without a fixed structure.

The RDBMS stores structured data. NoSQL databases can store semi-structured and unstructured data as well.

4.3.2 No fixed schema: JSON documents and key-value hierarchies

There is no fixed schema. Each record could have different attributes. There could be JSON documents, some key-value pairs, or a hierarchy of key-value pairs — as is the case with MongoDB. It is not the case that every row has the same set of attributes; that is simply not the case with NoSQL databases.

// Record 1 — a book
    { "title": "Big Data and Analytics", "author": "Seema Acharya", "year": 2011 }

    // Record 2 — a book that also carries an extra field the store never knew about
    { "title": "Big Data, Data Mining, and Machine Learning", "author": "Jared Dean", "year": 2014, "edition": 2 }

4.3.3 What NoSQL gives up: joins, transactions, and ACID

The NoSQL databases are non-relational in nature, so they do not support joins — typically. If you want to do joins, the RDBMS is the good choice. Some of the things can be done in NoSQL, but joins are very expensive if you do them with NoSQL databases. They are not suited for transaction semantics over multiple data items, and there is no support for ACID as in the RDBMS. The consistency semantics are relaxed. In some cases — the graph databases — the data can be modeled as graphs and queries written as graph traversals, which are nowadays getting a little more popular, specifically Neo4j.

4.4 The Four Families of NoSQL Databases

There are typically four kinds of NoSQL databases: key-value, document-oriented, columnar, and graph. We discussed these a little earlier; here is the formal classification.

When to pick which: if your scenario is "look up one thing by its identifier" choose key-value; "search inside records with varying fields" choose document; "run statistics over one column of a huge table" choose columnar; "follow relationships between entities" choose graph. The professor's rule stands: the choice is scenario-driven — no single storage engine is right for everything.

4.4.1 Key-value stores

A key-value store is like a big hash table: it maintains a big hash table of keys and corresponding values. Real-world: Examples of key-value pair databases include Dynamo, DynamoDB, Redis, and a fourth name transcribed as "React" — most likely Riak. Riak is in fact a well-known open-source key-value store that auto-distributes and replicates data across nodes, which fits this list exactly.

4.4.2 Document-oriented databases

A document-oriented database stores data in the form of documents — a collection of documents. Real-world: MongoDB is one of the very good examples of a document-oriented database, and CouchDB is another.

4.4.3 Column-oriented databases

A column-oriented database stores data in storage blocks. This is the HBase story: the concepts of column families and blocks we discussed earlier are exactly what column-oriented databases use. Take a table with attributes \(A_1, A_2, A_3, A_4\). Normally, all the values of one column are stored together, so that whatever analytical query runs on one particular column runs much faster — the column's data is one sort of a block, and once you access one value, the whole block is brought into main memory. That is the locality of reference idea. Queries that require a bunch of values from one column — mainly statistics on that column — run faster.

Additionally, HBase-style systems let you build a column family: combine two attributes, say \(A_2\) and \(A_3\), and store those two attributes together in sequence. Any query that needs access to both \(A_2\) and \(A_3\) at the same time will run faster. Real-world: Cassandra and HBase are the two most popular column-oriented databases. HBase's storage (blocks, column families) was already covered; Cassandra is covered in detail in this session.

4.4.4 Graph databases

There could be scenarios where your database is very, very connection-heavy — social media applications like Facebook and Instagram. The classic query: how many hops is this particular person away from that particular person? If you store this in an RDBMS, it becomes very, very difficult to query and retrieve the data, because one record of the database may be associated with multiple relationships — a person has written \(N\) number of posts, has commented on \(M\) number of posts, has liked \(X\) number of posts. Such cases can be modeled with graph databases, and Real-world: Neo4j is one of the popular examples.

4.4.5 The friend-of-a-friend problem

Walk through the hop-count example in detail. There is a friend list of person \(P_1\). \(P_1\) is a friend of \(P_2\), \(P_2\) is a friend of \(P_3\), and \(P_3\) is a friend of \(P_4\) — but \(P_1\) and \(P_4\) are not in direct link, and neither are \(P_1\) and \(P_3\). I want to figure out the friends of friends of \(P_1\). In a relational database this requires walking relationship joins repeatedly, and with a huge user base it becomes a bottleneck. The graph model makes this traversal natural — you follow edges from one node to the next.

4.5 Characteristics, Pros and Cons, and SQL versus NoSQL

4.5.1 Scale-out architecture and the three scales

All NoSQL families follow a scale-out architecture. There are two kinds of scaling: vertical and horizontal. Horizontal scaling is your scale-out: a cluster spread across 100+ nodes, across data centers also. The NoSQL world makes three concrete scale promises:

• Performance scale — support 100,000+ (one lakh, \(10^5\)) database reads and writes per second.
• Storage scale — store 1 billion+ (\(10^9\)+) documents in your database.
• Data variety — house large amounts of structured, semi-structured, and unstructured data.

4.5.2 Auto-sharding, replication, and failure recovery

There is no predefined schema, so data can be inserted without any predefined schema, and you can add tags — a JSON document has certain tags, and you add a tag and a corresponding value, a kind of key-value pair. The systems also follow auto-sharding. We talked about partitioning: you partition your data and give it to different nodes; the more partitions, the more jobs can be done in parallel, and the more performance. The same thing happens here: many of these databases automatically spread the data across a number of servers. Partitioning means they are sharding your data — horizontal partitioning — and even the applications are not aware of it; it happens under the cover. Partitioning helps in data balancing and failure recovery: if one partition fails, we may keep replications, and we can go to the other replicated data and get the results. This gives good support for high availability as well as fault tolerance. These are the characteristics of all NoSQL databases.

4.5.3 The pros of NoSQL

The pros: cost-effective for large data sets; easy to implement; easy to distribute, especially across data centers; easier to scale up and down; relaxes data consistency; no predefined schema.

4.5.4 The cons of NoSQL

The cons are the other side of the same coin:

• Does not support joins efficiently among the data sets in the tables — if you want to perform joins, go for the RDBMS.
• Group by operations are very costly in NoSQL databases.
• No support for ACID properties — if you want highly consistent data, NoSQL is not a good choice. It is not simply one blanket reason: configurations exist. We can configure MongoDB from different consistency levels, but as you increase the consistency, the performance will go down.
• Lack of standardization in this space, which makes it very difficult to port from SQL across NoSQL data stores.
• Skillset — people skilled in these technologies are fewer compared to SQL.
• BI tool support is less — fewer business intelligence tools can be integrated with these technologies compared with the mature BI space available with the RDBMS.

4.5.5 SQL versus NoSQL, side by side

Real-world: The vendors offering these different kinds of databases include Amazon, Facebook, Google, and Oracle — a mix of the companies that built the technologies and the traditional database players.

4.5.6 The CAP theorem revisited

The CAP theorem is very, very important. It was discussed before the mid-semester: Consistency, Availability, and Partition tolerance, of which any distributed system can deliver only two at a time. Partition tolerance becomes mandatory when your application is hosted across data centers, because network partitions are bound to happen. So \(P\) is fixed; out of \(C\) and \(A\), you have to go with one — either consistency or availability — and that choice depends on your application. That is exactly how you choose your database: the SQL side bets on consistency; the NoSQL side makes the C-versus-A trade-off explicit.

Exam note: Expect the CAP theorem and the consistency levels to keep returning in exam material — know the three letters, the trade-off, and why partition tolerance is not optional in distributed deployments.

4.6 Document-Oriented Databases in Detail

4.6.1 Documents, collections, IDs, and indexes

A document-oriented database stores data in the form of documents, typically JSON. The documents are accessible via their ID, and can be accessed through their index as well — MongoDB supports creating indexes on document-oriented data. The database maintains data in collections of documents. The mapping to the relational world is direct: a record in an RDBMS is equivalent to one document, and one table is equivalent to a collection. A collection has multiple documents in it, and you add documents to a collection the way you add records to a table — the insert operation works with document-oriented databases such as MongoDB.

4.6.2 A book document and querying by tags

Here is one sort of book document: book title, publisher, year of publication. You can store this document, say in MongoDB, and a unique document ID (an object ID) will be generated automatically. If I want to query something on book title, I can do that; if I want to query on year of publication, I can do that. Because each attribute is stored as its own tag, a search can be performed on different attributes.

{
      "_id": ObjectId("507f1f77bcf86cd799439011"),
      "book_title": "Big Data and Analytics",
      "publisher": "Wiley India",
      "year_of_publication": 2011
    }

4.6.3 Key-value store versus document-oriented: the use-case question

4.6.4 Searching with find

For searching, MongoDB provides a find function: you specify your search criteria there, and it filters the records according to whatever documents satisfy that search criteria, and displays them as part of the result. The CRUD operations — creating data, inserting, deleting, updating — will be covered in the MongoDB practical (about 30 minutes), and a dedicated session can go deeper: what a document is, how we store it, how we query, how we search, how we update.

4.7 Column-Oriented Storage: Laying Out Columns

4.7.1 A relational table with five columns

Look at a typical relational table with five columns: employee number, department id, hire date, last name, first name. There are unique employee numbers, the department ids in which the employees work, their hire dates, their last names, and their first names. The table may hold five records or many hundred thousand records.

4.7.2 Vertical partitioning into column families

In column-oriented storage, the data is stored in a different layout: a whole bunch of values of one column are stored together — all employee numbers together, then all department ids together, then all hire dates together. This partitions the table by column into column families — a kind of vertical partitioning — where each column family is stored in its own file. Each storage block has data from one column: this will be one storage block, that will be another storage block. The systems also allow the versioning of the data values within a block.

4.7.3 Building column families around query patterns

You can go beyond a one-column store and form a column family by clubbing columns, like in HBase: club employee number and department id, and store them together — (1, 1), (2, 1), (3, 1), (4, 2), and so on — because every time I refer to the employee number, I need the department id as well. Those queries become much faster. So you need to frame the column families carefully, looking at how the data will be queried — the family structure should mirror the access pattern.

4.8 Graph Databases: Modeling Connections

4.8.1 Nodes, edges, and properties

Graph computing — flagged during the discussion as one of the most important topics — deals with data represented in the form of graphs. A graph consists of nodes (also called vertices) and edges: the edges connect different nodes. Data is represented as vertices and edges. The vertices have certain properties — key-value pairs — and the edges represent the relationships, and edges can carry properties too.

4.8.2 A social graph in practice

Real-world: Picture a social media site: Alice is a friend of Bob, Bob is a friend of Charlie, and Charlie is a friend of Devin. All of these are persons — one entity — and the friend label indicates the kind of relationship between the entities. This is easy to interpret as a graph. In the storage model: a node named Alice with age 18 and id 1 is attached to another node, id 2, named Bob with age 22; the edge says Alice knows Bob since a particular date — the relationship is mentioned on the edge, with its own properties.

Visual intuition — the Alice-Bob-Charlie-Devin graph: picture a row of four dots on paper, left to right: Alice, Bob, Charlie, Devin. Three arrows connect consecutive dots, each labeled KNOWS and each carrying a small property tag ("since 2021-05-14", "since 2020-02-09", "since 2019-08-30"). The landmark to notice: Alice's dot has exactly one outgoing arrow, while the chain itself is four hops long — a traversal from Alice to Devin must cross three edges, each step moving one dot right. The takeaway: the shape of the picture is the shape of the data — no separate lookup table is needed to answer "how far is Devin from Alice?", the answer is visible in the arrow chain itself. This direct adjacency is exactly what native graph storage preserves on disk (Section 4.11.1).

4.8.3 Graph query languages: Cypher and Gremlin

To query data stored as graphs, you traverse the nodes — how that works is covered with Neo4j. A graph database is also called a network database, because it is good for representing network-style data. Two graph query languages are supported: Cypher, typically used for Neo4j, and Gremlin, for Apache TinkerPop.

4.9 Cassandra: The Column-Oriented, High-Availability Database

4.9.1 Origins: Facebook, DynamoDB, and Bigtable

Cassandra was developed at Facebook, and it was built on the concepts of Amazon DynamoDB and Google Bigtable. You can see the two lineages in its design: the Dynamo side contributes the distribution and availability thinking; the Bigtable side contributes the column-family storage model.

4.9.2 The AP design and the peer-to-peer architecture

Cassandra follows an AP design in the CAP context. Recall: when a distributed database is hosted on multiple data centers, partition tolerance is a must because partitions are bound to happen, and the CAP theorem says we cannot guarantee all three properties at once — we can guarantee only two, and since \(P\) is mandatory, we choose one of \(C\) and \(A\). Cassandra chooses availability: it is a high availability database that can sacrifice consistency. It is high performance, high availability, with a compromise on consistency. So if you want high availability and performance, with a little compromise on consistency, you go ahead and use Cassandra.

Architecturally, each node acts the same — the nodes are peer-to-peer, symmetric nodes; there is no primary-secondary split. Column-oriented storage is built the same way we just saw: you build column families based on your requirement, and each column family is stored as a storage block. The system is also, at heart, a key-value store.

4.9.3 Keyspaces, column families, and columns

Cassandra's vocabulary: you create a keyspace — a keyspace is like a database. Within a keyspace you create a column family, which is like a table. Within a column family you create attributes — called columns — with their data types. The nesting is: keyspace → column family → columns.

4.9.4 The write path: commit log, memtable, and SSTable

How do reads and writes happen in Cassandra? There is a commit log that maintains the sequence of operations happening in the database. Writes are entered into the log sequentially, and then they are deemed successful. The write goes to a single node and is committed there — there are configurations for that, as with MongoDB. Further, the data is indexed and put into an in-memory data structure called the memtable. Whatever writes are happening live in the memtable, and then the memtable is flushed to disk as an SSTable. An SSTable is immutable — you cannot update it; it is an append-only thing. Partitioning and replication happen automatically — the sharding we discussed. A default replication factor of three (\(RF = 3\)) is maintained in your distributed application automatically.

4.9.5 The read path and consistency levels

For a read, the client can connect to any node and read the data. The consistency level decides when a read is returned. If I read from one node and return, my read performance increases. But if I set it so that "this many nodes in the cluster have the same value", it should read from that many nodes and then return — some performance goes down. The setting says how many replicas should contain the same copy before I say "this is the value I want to return".

Walk through a worked example with a seven-node cluster: I say that four nodes have the same copy before returning a read. If I am reading from a single node, it has to check three more nodes; if those three more nodes have the same copy, return that read value — otherwise don't. I can also configure it the other way: whatever read value I have, I return it. So the read behavior depends entirely on the consistency you have configured. The same logic applies to writes: whether you want the write to be successful as soon as it hits one particular node, or you want a certain number of nodes to successfully write this update before confirming that the write is successful.

There is also read repair: if the data at a particular node is corrupted, there is a gossip protocol that can be leveraged to repair the corrupted things. That protocol is not a part of the course syllabus, but you could go through it once — Google it — to see exactly what it is.

4.9.6 Quorum consistency

You can set the consistency level to \(1, 2, 3\), up to any number \(n\) — the number of nodes that must have the same value before the write is considered successful. Write consistency \(1\): as soon as the write hits one node, the write is considered okay. But what if that single node dies? That is where consistency is compromised — and that is exactly why we call this an AP database. You can also configure the other extreme: until \(n\) number of copies all have the same value, the write is not considered successful — and then performance goes down. The same \(1..n\) choices apply to reads.

There is also the quorum option: when the majority of the nodes in the cluster have the same value, then the read or write can succeed. When the majority of nodes have been successfully written, the write is successful. So you can follow quorum consistency rather than a hard number \(n\) — the standard, balanced choice.

4.9.7 Lightweight transactions

Cassandra offers some lightweight transactions. Here is the interface with two commands — insert and update — with special clauses:

INSERT INTO cycling.cyclist (id, last_name, first_name)
    VALUES (5, 'VOS', 'Marianne') IF NOT EXISTS;

    UPDATE cycling.cyclist
    SET first_name = 'Marianne'
    WHERE id = 5
    IF first_name = 'Marianne';

The insert checks whether such a record already exists before inserting. The update changes the first name only if the current first name matches the condition. You can explore more on the linked documentation.

4.9.8 Replication strategies across data centers

The replication strategy for your user data is configurable. You can specify a replication factor of \(n\): the data is stored in \(n\) nodes of the cluster. It is not only the default of three — I can say four as well. In a network topology strategy, you can specify different replication strategies for different data centers: a data center that is heavily used may want higher replication for a certain kind of data; a data center that is not heavily used may not want a very high replication factor. This kind of query is possible: within the east data center, go with replication factor 2; within the west data center, go with replication factor 3.

4.9.9 Partitioners: hashing the data across nodes

Data is partitioned — partitioned based on hashing — to distribute the data blocks from a column among the nodes. First, this is column-oriented: the data of one column is stored together. Second, the distribution is not a naive division — it is not "pick the first 20 rows for node 1, the second 20 rows for node 2". You apply a hash function to distribute the data evenly across the different blocks and nodes. There are different techniques, in two categories:

• Crypto hash (for example, MD5) — very difficult to reverse, very, very secure, but much more expensive.
• Non-crypto consistent hash — relatively easier to reverse compared to the crypto hash, however this is faster, and you can have about a 10% (\(10\%\)) performance improvement over the crypto hash.

These are the partitioners available; we will not go into the details of what exactly they are, but this is the way partitioning works — that is what we need to know.

4.9.10 Sample Cassandra queries

Some sample queries to see the shape of the language:

CREATE KEYSPACE demo WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};

    DESCRIBE KEYSPACES;

    USE demo;

    CREATE TABLE student (id int PRIMARY KEY, name text);

    DESCRIBE TABLE student;

CREATE KEYSPACE creates the keyspace with a replication strategy and factor. DESCRIBE KEYSPACES is like describe databases or describe tables in SQL. USE demo selects the keyspace — your database. CREATE TABLE student creates a table; it is very, very similar to SQL, but not exactly SQL — people who know SQL will find it simpler. DESCRIBE TABLE shows the table definition.

4.9.11 eBay: Cassandra at marketplace scale

Real-world: A major case study of Cassandra is eBay. eBay is a marketplace with 100 million active buyers, 200+ million items, around 2 billion page views, 80 billion database calls, and multi-terabyte storage capacity. That is the volume of data being dealt with. There is no transactional data, no joins, no integrity constraints of the key kind. It is a multi-data-center deployment with 400+ million hits and 200+ million leads.

What is happening: social signals on product pages, where read latency is not important, but write performance is the key. So the basis of tuning is: set the consistency level. If I want to increase write performance, I lower the number for write consistency; if I want to set read performance high, I lower the number I set for read consistency. It all depends on the configurations and the kind of scenario you are looking for. Use cases include connecting users and items via buy, sell, bid, watch events, and many time-series algorithms running to detect the frauds that may be happening across a certain period of time. This is one example of Cassandra being used at marketplace scale; the slide deck for the case study has more detail.

Exam note: For Cassandra, know the write path (commit log → memtable → SSTable, immutable and append-only), consistency levels and quorum, and the \(RF = 3\) default — these recur in exam material.

4.10 MongoDB: The Document-Oriented Database

4.10.1 Collections, documents, and the 16 MB limit

A MongoDB database is a set of collections. A collection corresponds to a table in the RDBMS, and it holds the documents. Documents are JSON documents with hierarchical key-value pairs — similar to rows in a table — and they can also be binary JSON (BSON) documents. The maximum document you can store is 16 MB (\(16\,\text{MB}\)), in the WiredTiger storage engine. MongoDB uses two different kinds of storage engines: WiredTiger and GridFS. With WiredTiger you can store documents up to 16 MB; for larger ones you need to use GridFS. GridFS also supports binary data — you want to store large objects, very large images, and all those things — and those are stored in chunks of 255 KB (\(255\,\text{kB}\)), with the metadata stored in a separate collection. GridFS does not support multi-document transactions, and it can be queried from WiredTiger and other engines. So the summary: data is stored in the form of documents, collections hold multiple documents, the documents can be JSON or binary JSON, and two engines cover small and large documents.

4.10.2 Sharding: horizontal partitioning in practice

The important concept that MongoDB uses for scaling is sharding — nothing but horizontal partitioning: data is partitioned across multiple servers. Why does horizontal partitioning give horizontal scaling? As mentioned, the more the number of partitions, the more parallel jobs you can run. Take MapReduce as an example: if I have more partitions, more map jobs can run in parallel — if I want to perform a similar kind of job, all map jobs can run in parallel. So if my data has an appropriate sharding, more jobs get done in parallel and more performance is achieved. Second, it is good for horizontal scaling: you add a node to the cluster and your capacity increases. You perform re-sharding because that node has to be allocated a certain share of data, which reduces the amount of data each node handles as the cluster grows. If your data stays the same and you add a node, the amount of data each node handles goes down, and the number of operations each node has to manage also goes down, because the data is divided. The data is replicated as well.

4.10.3 Primary-secondary architecture and read preference

MongoDB is not peer-to-peer like Cassandra: it uses a primary-secondary configuration — what you can also call a master-slave kind of thing. Performance settings can be used to tweak the write consistency, the way we discussed in Cassandra: a number of nodes is set — when you go for a write, how many nodes should be written before you consider the write successful. Data is replicated, and the performance can be tweaked. Clients usually read the data from the primary, but the read performance setting can also be tweaked from the read consistency: do you want to read from a single node and consider it successful, or read from a majority of nodes and then consider it successful? One difference from Cassandra: here the data is not time-stamped or versioned the way it was done with Cassandra — the data is used as it is.

4.10.4 Indexes, sub-document fields, and a sample document

Some form of indexing can be performed in MongoDB: you can create an index on any field of a collection, or on a sub-document field. Here is a sample document. The document has a name, Ravi, and an address field, which is again a hierarchy: city, state, pin code. And there are tags — a collection of tags: football, cricket, badminton. This is a sort of key-value pair structure — one document — and you can store this as one record of MongoDB. Then you can query: "I want to figure out the records with the city New Delhi" — this document will appear as part of your search. Further, you can create indexes and use the find function to search by them. Creating an index on tags looks like this — and now, searching on "tags: cricket", this may speed up the performance a little bit. Inserting data, deleting data, and updating data will all be covered in the lab; creating an index is also something we can look into during the lab.

{
      "_id": ObjectId("5f8b4d2c1a2b3c4d5e6f7081"),
      "name": "Ravi",
      "address": {
        "city": "New Delhi",
        "state": "Delhi",
        "pin_code": 110001
      },
      "tags": ["football", "cricket", "badminton"]
    }

4.10.5 Joins with aggregate and the lookup stage

This slide indicates that in MongoDB 3.2 plus it is possible to join the data from two collections using the aggregate pipeline. Look at the collections: this is the collection books — one collection, one "table" — with three attributes: isbn number, title, author. And there is a book selling data collection with isbn number and copies. Using db.books.aggregate we can perform a kind of join — the lookup stage (written with a dollar sign before its name):

db.books.aggregate([
      {
        $lookup: {
          from: "books_selling_data",
          localField: "isbn",
          foreignField: "isbn",
          as: "copies"
        }
      }
    ]);

We take the books collection, look into the books_selling_data collection, match the local field isbn against the foreign field isbn, and collect the matches under the field "copies". People who know SQL know how to join two tables — join on isbn with isbn and aggregate the data; the same thing is happening here. This is the direct join in MongoDB. Although it is not preferable to use NoSQL databases where frequent joins are being used, this is one support that is there in MongoDB 3.2 plus.

SELECT * FROM books LEFT JOIN books_selling_data
    ON books.isbn = books_selling_data.isbn;

4.10.6 Read and write concerns, and causal consistency

MongoDB offers various read and write choices for flexible consistency trade-offs with scale, performance, and durability — the read consistency and write consistency we have talked about can both be set, based on what performance and durability you are looking for. In case of a network partition, because this is a primary-secondary architecture, it may elect another primary — re-election on primary failure. The client application reads and writes from the primary; writes are communicated to the secondaries, and then the write can be considered successful. Reads can happen from the primary. If I want to increase read performance, I read from the primary. The read can also come directly from a secondary — like that arrow — if the primary is too busy; you read from the secondary for load-balancing. And if the primary dies, an election happens among the secondaries; one of them acts as the primary and starts its operations.

The read concern options: local means the client reads from the primary replica; then there are the causal-consistent-session options and available; and majority — the client wants to read what the majority of nodes have. Majority is the best option for fault tolerance and durability. The same idea as quorum in Cassandra: in Cassandra the term is quorum, in MongoDB the similar term is majority. The write concern options: write to one, write to \(n\), or write to the majority. If you want performance, go with the lower options; if you want durability, go with majority.

The standard setup uses an odd number of nodes in a cluster. If there is a network partition or some failure where the nodes are divided into two, one partition will always have a larger number of nodes than the other. If I set read to majority and write to majority, my system gives me causal consistency. Whenever I read, I read from the majority of nodes; only when the majority of nodes have the same copy do I consider the read successful. When the majority of nodes have the new update written, I consider the write successful. So with a majority of the nodes holding the same value, the system is causally consistent — this was discussed earlier as well, along with the eventual consistency scenarios.

Here is the causal-consistency recap, with four processes \(P_1, P_2, P_3, P_4\) on a timeline: \(P_1\) has written a value of \(x\) to 5 at this moment; \(P_2\), when it reads, reads \(x\) as 5; \(P_3\) reads \(x\) as 5 and later reads \(y\) as 10; and \(P_4\) reads \(y\) as 10 but reads \(x\) as 0 — \(x\) was updated earlier. \(P_4\) sees a write to \(y\) but not the write to \(x\). That schedule is not causally consistent — and it is also not linearizable, not strict, and not sequential. For each of these consistency levels — linearizable, strict, sequential, eventual — you can refer to the previous coverage; there is also a link in the earlier slide deck for reading more.

Exam note: Expect the consistency levels and read/write concerns to keep appearing in exam material — know the MongoDB read/write concerns (local, available, majority; 1, n, majority) and that majority is the option for fault tolerance and durability.

4.11 Graph Computing: Neo4j and Apache TinkerPop

4.11.1 Native versus non-native graph storage

There are two kinds of graph storage. Native storage stores the data in-house — the database itself manages the storage. The popular one is Neo4j, which maintains its own data storage. Non-native storage — the graph computing platforms — makes use of external databases for storage, and on top of that provides a graph layer. Apache TinkerPop is a computing platform that connects to graph databases that actually store the nodes and edges. Its built-in TinkerGraph stores data in memory only — so this is not a database storage; it actually creates a layer on top of other database storages, such as Elasticsearch or any other kind of database that can store this data — and then you access that data in a graphical way using the Gremlin language. The native approach is much faster, because adjacent nodes and edges are stored closer to each other, which gives faster traversal. In a non-native approach, extensive indexing has to be used. The native approach also scales as nodes get added.

4.11.2 Why relational tables struggle with relationships

Why model this kind of scenario as a graph instead of a relational database? A relational database stores data in the form of tables. I can have a person table with four records — Alice, Bob, Charlie, Devin. How do I model "Alice is a friend of Bob"? I have to come up with a relationship of foreign keys, and there will be a self-referential foreign key — Alice is a friend of Bob, something like that. When you have a lot of this, it becomes very, very difficult to manage using the relational database. And this is only the friends case. What if we also need to monitor that Alice has posted 10 posts, has liked 50 posts, has joined these many groups? For each of these things I have to maintain one relational table, and Bob will be linked to 50 of the like records, and to however many groups he has joined. If I store this kind of normalized data and want to gather "how many of the posts were liked by Alice", I have to do a lot of joins. Joins are an expensive operation — and if there is a lot of data, like many users on a social platform, and I keep doing joins at the back end, all this will become a bottleneck. The solution is to model such scenarios as graphs: store the data in the form of nodes and edges, let the edges represent the relationships, and traverse the graphs based on whatever the requirement is.

4.11.3 When to choose a graph database

When do you use a graph database? Three signals from the discussion:

1. You have a very heavy relationship data set — a large set of data items where the connections between them matter most.
2. Your queries are graph traversals, and you need to keep the query performance almost constant as the database grows. With a relational DB, if I have 100 records in every table, the performance may be okay. But these kinds of use cases grow exponentially, and that results in a performance bottleneck in relational databases. You want: one more user comes to the platform, tries to connect — the performance should not go down.
3. There is a variety of queries that are asked from this data, and they almost touch upon each and every attribute. If my queries are touching each and every attribute, even creating indexes in a relational DB does not help — what if I have 100 attributes? Will I create 100 indexes? It becomes very cumbersome to manage. So: connection-heavy, relationship-heavy data, diverse queries touching every attribute, and performance that must hold as the database size grows — that is when graph computing or graph storage is a good choice.

4.11.4 Cypher: querying the movie database

Neo4j uses Cypher, a declarative query language — like SQL, but for querying the graph. There is a popular movie search database: a person node, Tom Hanks, who has acted in a certain number of movies; the edges indicate "acted in" — Cloud Atlas, The Da Vinci Code, Charlie Wilson's War, and others. The movie nodes have incoming edges — "acted in", "directed by" — and you see that Mike Nichols directed a particular movie. Let us build the queries step by step.

First, the simple query — find the person named Tom Hanks, follow the ACTED_IN edges, collect all the movies into a set \(m\), filter with a WHERE clause like in SQL, and return just five results:

MATCH (p:Person {name: 'Tom Hanks'})-[:ACTED_IN]->(m)
    WHERE m.released > 2000
    RETURN m
    LIMIT 5;

"Match person named Tom Hanks" — the person is one entity in the database. "Further, we are looking for the acted-in edges; we are looking for all set of movies, and that is being stored in set m. Why am I storing it? Because I want to apply filters on this set." The WHERE clause filters: movies released after the year 2000, and return five results — it may return 100 results, but I am looking for the first five. The query does not just return text: it returns a graph structure, and I can click on any of these nodes to further explore it — details about Cloud Atlas, or about this person — that is also a possibility.

Second, refine: find the movies that Tom Hanks acted in and that were directed by Ron Howard:

MATCH (tom:Person {name: 'Tom Hanks'})-[:ACTED_IN]->(m)<-[:DIRECTED]-(ron:Person {name: 'Ron Howard'})
    WHERE m.released > 2000
    RETURN m
    LIMIT 5;

This says: match a person named Tom Hanks who acted in movies set \(m\); then, with the comma, search for a person named Ron Howard who has directed certain movies; out of whatever set \(m\) was retrieved, look for any directed edge to these movies — an edge with this particular label — with the same condition \(m.released > 2000\), and return five results. So: movies Tom Hanks acted in, directed by Ron Howard, released after 2000. Directed is a relationship between a person and a movie; acted in is a relationship between a person and a movie — between the same two entity types there are multiple relationships. In SQL you would maintain two tables and put across joins; this is how it is done in the graph-traversal way.

Third, add one more condition: who were the other actors in the movies where Tom Hanks acted and Ron Howard directed?

MATCH (tom:Person {name: 'Tom Hanks'})-[:ACTED_IN]->(m)<-[:DIRECTED]-(ron:Person {name: 'Ron Howard'}),
          (p:Person)-[:ACTED_IN]->(m)
    WHERE m.released > 2000
    RETURN p
    LIMIT 5;

Till this point we have figured out the movie data set in which Tom Hanks worked. Then, with this comma, we figure out out of these movies how many are directed by Ron Howard. Now in the persons' database again, we search how many of them have acted in the movies we have found here — the set \(m\) is getting filtered, first here, second here, third here, with the condition \(m.released > 2000\). The result set is stored in \(p\), and we list the persons. These are very simple queries to start with; a lab session on Neo4j (along with MongoDB) will show how the graph looks, how we can explore the nodes that come as part of a query result, and we will load the movie search database into Neo4j and see what kind of database it is.

4.11.5 Gremlin: querying with TinkerPop

TinkerPop supports the Gremlin query language, which has both declarative and imperative flavors. The sample queries: movies where Tom Hanks has acted — this is simple; we have a person, we have a movie, and we want the ACTED_IN relationship.

g.V().hasLabel('person').has('name', 'Tom Hanks').out('acted_in').values('name')

Here we are starting with the whole graph — that is what \(g\) is. I am looking for a set of vertices, \(V\), which has a label person — because some set of vertices will also have the label movie. I want to look for persons further, which have a name Tom Hanks, and which have an outgoing edge acted_in; .values('name') gives all the names of those movies Tom Hanks has acted in.

Now the second query — Tom Hanks acted in and directed by Ron Howard. Again you start with \(g\), you go to the set of vertices, you filter out the vertices with the label person, you filter out the records of the person with the name Tom Hanks, and further filter where there is the outgoing edge acted_in with that label. Now this movie should have an incoming edge directed_by — an incoming edge which has a name Ron Howard:

g.V().hasLabel('person').has('name', 'Tom Hanks').out('acted_in')
      .in('directed_by').has('name', 'Ron Howard').values('name')

In the first query we are going from the person entity to the movie entity; in the second one, from acted_in we are going from the person entity to the movie entity and then further to the director person entity — the difference is that the first time we are looking for Tom Hanks, the other time we are also looking for Ron Howard. The lab will discuss more of these things — probably not on TinkerPop, but definitely on Neo4j, with queries we will talk about in the lab session.

Exam Guidance Summary

• Graph databases are one of the most important topics. This was flagged explicitly. Know the graph model (nodes, edges, properties), when to choose a graph database (relationship-heavy data, traversal queries with near-constant performance, diverse queries touching every attribute), native versus non-native storage (Neo4j native and fast because adjacent nodes and edges are stored close; TinkerPop/TinkerGraph as an in-memory layer over external stores like Elasticsearch), and the two query languages — Cypher (Neo4j, declarative, like SQL) and Gremlin (Apache TinkerPop, declarative and imperative flavors).
• CAP theorem — very, very important. It was flagged as very important and was discussed before the mid-semester too. Know consistency, availability, and partition tolerance; be able to say why partition tolerance is mandatory in a multi-data-center deployment and why you must choose between C and A. Cassandra is the AP example: high availability, high performance, sacrifices consistency.
• Consistency levels and configurations are exam-grade material. The causal-consistency example (\(P_1\) writes \(x = 5\); \(P_4\) reads \(y = 10\) but \(x = 0\) — sees the write to \(y\) but not the write to \(x\) — so not causally consistent, not linearizable, not strict, not sequential) is the classic question shape. Review the full hierarchy (linearizable, strict, sequential, causal, eventual) from the earlier coverage — the revisit in this session was brief on purpose because it was already covered in detail before the mid-semester. Expect the MongoDB read/write concerns (local, available, majority; 1, n, majority) and the Cassandra consistency levels and quorum in the same family of questions. Remember: in MongoDB the option is called majority; in Cassandra the same idea is called quorum.
• NoSQL classification and characteristics. Four families — key-value, document-oriented, column-oriented, graph — with examples (Dynamo, DynamoDB, Redis, Riak; MongoDB, CouchDB; Cassandra, HBase; Neo4j). Be ready for the SQL-versus-NoSQL comparison (schema, scaling, ACID versus CAP, joins, vendor support) and the pros/cons list.
• Cassandra internals. Write path: commit log → memtable → SSTable (immutable, append-only); default replication factor of three, configurable per data center (east 2, west 3, for example); partitioners — crypto hash (MD5) versus non-crypto consistent hash (~10% performance advantage); consistency levels 1..n and quorum; lightweight transactions and the IF clause; the seven-node worked example (four nodes must hold the same copy before a read returns).
• MongoDB internals. Collections and documents, JSON and BSON, the 16 MB document limit in WiredTiger and GridFS chunks of 255 KB for larger objects; sharding as horizontal partitioning and how more partitions enable more parallel jobs (the MapReduce analogy); primary-secondary architecture and re-election; indexes including sub-document fields; the lookup join (MongoDB 3.2+).
• Rack-allocation question from the mid-semester (makeup exam). A file was given stating that a certain number of blocks exist, and you had to draw a diagram of how you would place these blocks into different racks. The rules came from a slide discussed earlier: one copy is placed on one node in one rack, the second copy in another rack, and so on, with rules on how many replicas can sit on a particular rack; if a placement satisfies the rules, you can place it immediately. There could be multiple correct answers — you were supposed to follow the rules, not a particular algorithm. One node can have multiple blocks, but you do not put the same block on the same node multiple times. The exact replica-per-rack rules live on the shared slide from the earlier session — review that slide before the exam, because the question tests the rules, not a single prescribed layout.

Key Industry Applications

• Facebook — developed Cassandra, built on the concepts of Amazon DynamoDB and Google Bigtable.
• eBay — runs Cassandra at marketplace scale: 100 million active buyers, 200+ million items, about 2 billion page views, 80 billion database calls, multi-terabyte storage, multi-data-center deployment, social signals on product pages where write performance matters and read latency does not; consistency-level tuning decides write versus read performance.
• Amazon — the shopping-cart failure anecdote (adding an item sometimes fails) as a case where strict consistency is not required; the Dynamo and DynamoDB key-value stores.
• Google — Bigtable (Cassandra's storage lineage) and its own NoSQL technologies at the naming conference; Yahoo likewise contributed technologies to the movement.
• Oracle — the traditional vendor side of the SQL-versus-NoSQL comparison (excellent vendor support on the SQL side; community support on the NoSQL side).
• OnePlus — flash-sale website crashes from a few years back as the motivating example of burst traffic and the need for throughput without bottlenecking.
• Redis, Riak, Dynamo, DynamoDB — key-value stores; MongoDB, CouchDB — document-oriented; Cassandra, HBase — column-oriented; Neo4j — graph.
• Neo4j — the movie search database (Tom Hanks, Cloud Atlas, The Da Vinci Code, Charlie Wilson's War, Ron Howard, Mike Nichols) queried with Cypher.
• Apache TinkerPop / TinkerGraph / Gremlin — the in-memory graph layer over external storages such as Elasticsearch, queried with Gremlin.
• WiredTiger and GridFS — MongoDB storage engines for documents up to 16 MB and larger objects in 255 KB chunks.
• MD5 — the crypto hash used in Cassandra partitioning; non-crypto consistent hashing as the faster alternative with about a 10% performance edge.