Designing Data-Intensive Applications

This page is meant to act as a summary of notes I took while reading Designing Data-Intensive Applications by Martin Kleppmann. Specifically, the first edition.

None of the text on this page was AI generated.

I read the book, annotated as I went along, and wrote down notes in a Google Doc. This page is the web-friendly version of that Google Doc, created through much formatting and effort.

Much of the text here is either directly copied from Kleppmann’s own writing, my own summary/rewording, or some combination of the two. I also pull in information from other sources occasionally, but I try not to stray too far from Kleppmann as my source of truth. In the cases where I directly pilfer Kleppmann’s own writing, my aim is to provide a substantially different reading experience here. Similar ideas and concepts are usually formatted for easier comparison through tables or other visualizations. This formatting is directly from my own notes. No AI told me what to put in a table and where.

You will ocassionally find empty table cells in this guide. This is deliberate. If I am not 110% sure as to the answer the cell should describe, or if Kleppmann’s own writing on the subject seems ambigous, I defer to leaving that area empty.

This page is not meant to act as a replacement for reading the book itself, especially since a second edition is out now. Think of this more like a review/cheat sheet. Cheers!

Response Time

Measuring Response Time

The distribution of values should be thought of in percentiles. For example, if the 95th percentile response time is 1.5 seconds, that means 95 out of 100 requests take less than 1.5 seconds. High percentiles of response times, also known as tail latencies, are important because they directly affect a user’s experience of the service. For example, Amazon describes response time requirements for internal services in terms of the 99.9th percentile, even though it only affects 1 in 1,000 requests.

If an end-user request requires multiple back-end calls, a higher proportion of end-user requests end up being slow (an effect known as tail latency amplification).

On the other hand, optimizing the 99.99th percentile (the slowest 1 in 10,000 requests) was deemed too expensive and to not yield enough benefit for Amazon’s purposes. Reducing response times at very high percentiles is difficult because they are easily affected by random events outside of your control, and the benefits are diminishing.

Testing Response Time

It is important to measure response times on the client side. When testing, the load generating client needs to keep sending requests independently of the response time. If the client waits for the previous request to complete before sending the next one, that behavior has the effect of artificially keeping the queues shorter in the test than they would be in reality, which skews the measurements.

SQL Vs NoSQL

Declarative vs Imperative

function getSharks() {
    var sharks = [];
    for (var i = 0; i < animals.length; i++) {
        if (animals[i].family === "Sharks")
            sharks.push(animals[i]);
    }
    return sharks;
}

SELECT *
FROM animals
WHERE family = 'Sharks'

Document-Oriented Databases

The document in “document-oriented” can refer to JSON, XML, YAML, and BSON. Databases include MongoDB, RethinkDB, CouchDB, and Espresso.

A foreign key in a relational model is analogous to a document reference in a document model.

Document-oriented databases are good when you need large parts of the document at the same time. This avoids needing multiple lookups across different parts of the database. They target use cases where data comes in self-contained documents and relationships between one document and another are rare.

Graph-Like Data Models

Graphs are good for evolvability: as you add features to your application, a graph can easily be extended to accommodate changes in your application’s data structures. Queries to graph databases are partially characterized by having mechanisms for an unknown number of joins, since you don’t know how many nodes and vertices you need to traverse to retrieve your data. No schema is technically enforced.

Indexes

In order to efficiently find the value for a particular key in the database, we need to use an index. The general idea behind them is to keep some additional metadata on the side, which acts as a signpost and helps you locate the data you want. Any kind of index usually slows down writes, because the index also needs to be updated every time the data is written. However, well-chosen indexes speed up reads.

Hash Index

One kind of index is a hash index. Throw the key into a hash function and that’s your index!

• The hash table must fit in memory. You could theoretically maintain a hashmap on disk, but it’s very difficult for it to perform well due to random access I/O. This is bad for durability.
• Range queries are not efficient. You have to look up each key individually.
• O(1) reads and writes in memory

B-Tree Index

A self-balancing tree on disk.

• No biggie if the hardware dies while we’re updating the B-Tree, that’s why we typically use a WAL with B-Trees
• Since it’s stored on disk, it’s not as fast as say, a hash index that is stored on memory
• Relatively good for reads since you only need O(logn) for key retrieval, since that’s the height of the tree
• Writes can be slower since it’s on disk
• Big datasets are way more feasible since data is stored on disk, and has more space
• Range queries are easy to do, since keys are sorted and co-located

Log-Structured Merge-Tree (LSM Tree)

An LSM Tree is a 2-part mechanism for storing indexes: the memtable and the SSTables.

Memtable

The first part is the tree, which is a balanced tree (red-black tree, AVL tree, many options). This tree is sometimes known as a memtable. As the name implies, it is stored in memory.

When the memtable gets bigger than some threshold (typically a few MB), it’s written out to disk as a Sorted String Table.

Sorted String Table (SSTable)

A sorted list of key-value pairs flushed out from the memtable. This is easy to put in sorted order because the source memtable is a balanced tree.

Merging and Compaction

SSTables are append only. You will eventually get to a state where you have multiple SSTables, with duplicate values in different (or even the same) data segments. A background process will merge and compact these SSTables into a single one at regular intervals. This process throws away duplicate keys, keeping only the most recent version of each key. If a key is deleted, a special value known as a tombstone is inserted to represent its deletion. If the last known value for a key is a tombstone, all of its older values are purged in the merge and compaction process.

Key Lookup

To look up a key, you check the memtable first, then all the SSTables. This is costly, especially if the key does not exist. For this reason bloom filters are commonly used to speed up this process. Through a combination of hash functions and manipulation of a bit string, bloom filters can tell you if a key does not exist in a database, and thus saves many unnecessary disk reads for nonexistent keys.

• Supports range queries, but slow compared to B-Trees, since all SSTables need to be checked
• Extra CPU usage for compaction
• Number of keys not limited to memory

B-Trees vs LSM Trees

Multi-Column Indexes

The most common type of multi-column index is called a concatenated index, which simply combines several fields into one key by appending one column to another. This is like an old-fashioned paper phone book, which provides an index from (lastname, firstname). Last names are sorted first, then first names.

Found The Key, But Where Is The Value?

The key in an index is the thing that queries search for, but the value can be one of three things:

1. The actual row in question (clustered index)
2. A reference to the row stored elsewhere (nonclustered index)
3. A reference to the row stored elsewhere, along with some data from the actual row (covering index)

In nonclustered and covering indexes, the location where the rest of the row data lives is known as a heap file. There is a performance penalty for navigating to a heap file. Clustered and covering indexes can speed up reads, but they require additional storage and can add overhead on writes.

In-Memory Databases

In-memory key-value stores. Some are intended for caching use only (Memcached). Others aim for durability, achievable by:

• Battery-powered RAM
• Writing a log of changes to disk
• Writing periodic snapshots to disk
• Replicating the in-memory state to other machines

Examples include VoltDB, MemSQL, OracleTimesTen, RAMCloud, and Redis.

Counterintuitively, the performance advantage of in-memory databases is NOT due to the fact that they don’t need to read from disk. Even disk-based storage engines may never need to read from disk if you have enough memory. Rather, they can be faster because they can avoid the overheads of encoding in-memory data structures in a form that can be written to disk.

Transaction Processing & Analytics

OLTP vs OLAP

OLAP (Online Analytics Processing) is done on a data warehouse, which is a separate database that analysts can query without affecting OLTP (Online Transaction Processing) operations. The process of getting data into the warehouse is known as Extract-Transform-Load (ETL).

Many data warehouses are used in a fairly formulaic style, known as a star schema (or dimensional modeling). At the center of the schema is a so-called fact table. Each row represents an event that occurred at a particular time. Some of the columns will contain attributes of the event, while others will contain foreign key references to dimension tables. The dimensions represent the who, what, where, when, how, and why of the event.

A variation of this template is known as the snowflake schema, where dimensions are further broken down into subdimensions.

Column-Oriented Storage

If you have trillions of rows and petabytes of data in your fact tables, storing and querying them efficiently becomes a challenging problem. Although fact tables are often over 100 columns wide, a typical data warehouse query only accesses 4 or 5 of them at one time. Loading all the other column data for the rows you’re querying from disk into memory, parsing them, and then filtering them, is expensive. That is why data warehouses frequently rely on column oriented storage.

Compression

Depending on the data in the column, different compression techniques can be used. One technique that is particularly effective in data warehouses is bitmap encoding. Often, the number of distinct values is small compared to the number of rows, as there are usually repeat values. We can take n distinct values and turn them into n separate bitmaps. Each bit represents one row. The bit is 1 if the row has that value, and 0 if it doesn’t.

We can further compress these bitmaps with run-length encoding, which works like so:

10000001100 → 1, 6, 2, 2 (1 one, 6 zeros, 2 ones, 2 zeros)

Bitwise AND and OR operations can be used to operate on such chunks of compressed column data directly. This technique is known as vectorized processing.

Sort order can also help with column compression. If the primary sort column does not have many distinct values, then after sorting, it will have long sequences where the same value is repeated many times in a row. Run-length encoding can be used to compress that column.

Caching

A materialized view is a database object that represents the result of some query. They are not often used in OLTP databases because they are written to disk, and need to be updated when the underlying data changes. This makes writes expensive. However, in a read-heavy database, like OLAP, they make much more sense. They typically use a materialized view known as a data cube or OLAP cube. The advantage of this is that certain queries become very fast because they’ve effectively been precomputed.

Encoding Formats

Dataflow

RPC

A remote procedure call (RPC) is when a program makes a request to a remote network service look the same as calling a function in your programming language. This is not a great paradigm because it conflates local and remote execution, which are fundamentally different:

REST vs SOAP

The API of a SOAP web service is described using an XML-based language called the Web Services Client Description (WSDL). Features of WSDL:

• Enabled code generation, so that the client can access a remote service using local classes and method calls. This is useful in statically typed languages, and less so in dynamic ones.
• Not human-readable, need tools like IDEs to use

Async Messaging Passing Systems

One main difference is that message passing is usually one-way, unlike RPC.

Examples of modern message brokers include RabbitMQ, ActiveMQ, HornetQ, NATS, and Apache Kafka. They generally work as follows: One process sends a message to a named queue or topic, and the broker ensures that the message is delivered to one or more consumers or subscribers of that queue or topic.

Distributed Actor Framework

Replication means keeping a copy of the same data on multiple machines that are connected via a network. Each node that stores a copy of the database is called a replica.

Leader-Based Replication

1. One of the replicas is designated as the leader. When clients want to write to the database, they must send their requests to the leader, which first writes the new data to its local storage.
2. The other replicas are known as followers. Whenever the leader writes new data to its local storage, it also sends the data change to all of its followers as part of a replication log or change stream. Each follower takes the log from the leader and updates its local copy of the database accordingly.

Writes are only accepted by the leader (followers are read-only). Reads are accepted by both the leader and followers.

Sync vs Async Replication

Sync / Semi-Sync

In synchronous replication, the leader waits until the follower sends a success response back before informing the user/client of a success message.

Following from the “Con” column above, it is impractical for all followers to be synchronous: any one node outage would cause the whole system to grind to a halt. In practice, if you enable synchronous replication on a database, it usually means one of the followers is synchronous, and the others are asynchronous. If the synchronous follower becomes unavailable or slow, one of the asynchronous followers is made synchronous. This guarantees that you have an up-to-date copy of the data on at least two nodes. This is also known as semi-synchronous (since not all of the followers are synchronous).

Async

In asynchronous replication, the leader does NOT wait for a response from the follower. This is the most common configuration for leader-based replication.

Adding a New Follower

1. Take a consistent snapshot of the leader’s database at some point in time. Most databases have this feature anyways, as it’s required for backups.
2. Copy the snapshot to the new follower node.
3. The follower connects to the leader and requests all the data changes that have happened since the snapshot was taken.

Node Outage

Follower

On its local disk, each follower keeps a log of the data changes it has received from the leader. If a follower crashes, it can recover quite easily from its log. It can then connect to the leader and request all the data changes that occurred during the time when the follower was disconnected, and then continue as normal.

Leader

When a leader fails, a (usually automatic) process happens known as failover, which consists of the following steps:

1. Determining that the leader has failed (usually via timeout)
2. Choosing a new leader (can be done via controller node or election process)
3. Reconfiguring the system to use the new leader
   3.1. Clients need to send their write requests to the new leader
   3.2. Old leader becomes a follower and recognizes the new leader

Replication Methods

Statement-based replication

The leader logs every write request (statement) that it executes and sends that statement log to its followers.

Write-ahead log (WAL) shipping

Use a write-ahead log to build a replica on another node. The downside is that a WAL describes the data on a very low level, such as which bytes were changed in which disk blocks. This makes it closely coupled to the storage engine. This can mean that upgrading to a new version of the database software requires downtime.

Logical (row-based) log replication

To decouple the log from the storage engine, you can use a logical log. Rather than byte-level granularity, they use row-level granularity.

Trigger-based replication

This is replication that happens at the application layer. This is useful for when you need to replicate data to different database systems or custom conflict resolution logic. This approach is prone to bugs and incurs greater overheads. It works via a trigger, which lets you register custom application code that is automatically executed when a data change occurs in a database system.

Replication Lag

Replication lag is the delay between a write happening on a leader and when it is reflected onto a follower.

Read-after-write consistency

Read-after-write consistency is a guarantee that if a user reloads a page, they will always see any updates they submitted themselves. It is not guaranteed by default in a leader-based replication scheme. A write can be sent to a leader, who then forwards it along to a follower. However, a read request may have been sent to a follower before that could happen, to which they’ll retrieve old data (the newer data has only made its way to the leader). Methods for achieving read-after-write consistency with leader-based replication:

• When reading something that the user modified, read it from the leader; otherwise, read it from a follower. This only works if most things in the application are not editable by the user. If that is not the case…
• Use other criteria to determine whether to read from the leader. You could track the time of the last update, and for 1 minute after that time, make all reads from the leader.
• The client can remember the timestamp of its most recent write. Then the system can ensure that the replica serving any reads for that user reflects updates at least until that timestamp.

If your replicas are distributed to multiple data centers, you have additional constraints. Any requests that need to be served by the leader must be routed to the datacenter that contains the leader. This is especially applicable when implementing cross-device read-after-write consistency. This is the idea that if the user enters some information on one device and then views it on another device, they should see the information they just entered. A computer on home WiFi and a mobile phone on a cell signal will not necessarily route to the same datacenter. If your approach requires reading from the leader, you may first need to route requests from all of a user’s devices to the same datacenter.

Monotonic Reads

Monotonic reads means if a user makes several reads in sequence, they will not see time go backwards, i.e., they will not read older data after having previously read newer data.

One way of achieving monotonic reads is to make sure that each user always makes their reads from the same replica. This replica can be chosen based on a hash of their user ID perhaps.

Consistent Prefix Reads

Consistent Prefix Reads guarantees that if a sequence of writes happens in a certain order, then anyone reading those writes will see them appear in the same order. This is a particular problem in partitioned (sharded) databases. Different partitions usually operate independently, so there is no global ordering of writes. One solution is to make sure that any writes that are causally (not casually) related to each other are written to the same partition.

Leaderless Replication

Clients send writes to several nodes. This kind of database is known as Dynamo-style, based off of Amazon’s in-house Dynamo system. It made leaderless replication fashionable after a long period of obsolescence.

What happens when a node goes offline for a while, and then comes back online? It missed a considerable amount of data. There are two mechanisms that are often used to handle this:

Quorum

Separately, you can use a quorum: the minimum number of nodes that need to vote on an operation before it can be considered successful. This is determined via the following equation: w + r > n, where

• w is the number of nodes written to successfully
• r is the number of nodes read from successfully
• n is the total number of replicas

If this equation is obeyed, the reads and writes are called quorum reads and quorum writes. With quorum reads, one of the r replicas you read from must have seen the most recent successful write. Keep in mind, however, that quorums are not strongly consistent (reads made at the same time will not necessarily return the same value).

What do you do if, because of a network interruption, r or w nodes cannot be reached? Then a quorum cannot be determined. In this case, you can use a sloppy quorum. Writes and reads still require w and r successful responses, but those may include nodes that are not among the designated n “home” nodes for a value.

Once the network interruption is fixed, any writes that one node temporarily accepted on behalf of another node are sent to the appropriate “home” nodes. This is known as hinted handoff.

Version Vector

Replication Topology

• In circular and star topology, if just one node fails, it can interrupt the flow of replication messages between other nodes.
• All-to-all topology is the most general. Avoids a single point of failure so has good fault tolerance. However, some network links may be faster than others, with the result that some replication messages may overtake others.

Comparing Replication Approaches

The main reason for wanting to partition data is scalability. Different partitions can be placed on different nodes in a shared-nothing cluster. Our goal with partitioning is to spread the data and the query load evenly across nodes. If every node takes a fair share, then, in theory, 10 nodes should be able to handle 10 times as much data and 10 times the read and write throughput of a single node. If some partitions have more data or queries than others, we call it skewed. A partition with a disproportionately high load is called a hot spot.

Partitioning Of Key-Value Data

In this table, we only consider the case where we access these values by their primary index (the key). Accessing them with any other criteria would require a full scan. The following table compares various partitioning methods.

Partitioning and Secondary Indexes

Secondary Indexes do not identify a record uniquely. Rather, they are a way of searching for occurrences of a particular value.

This article is helpful for diving deeper into this topic with visual aids.

Rebalancing Partitions

The process of moving load from one node in the cluster to another is called rebalancing. Rebalancing works differently depending on how you are partitioning nodes.

Request Routing

When a client wants to make a request, how does it know which node to connect to? This problem is known as service discovery. There are different approaches:

• Allow clients to contact any node and make them handle the request directly, or forward the request to the appropriate node.
• Send all requests from clients to a routing tier first that acts as a partition-aware load balancer.
• Make clients aware of the partitioning and the assignment of partitions to nodes.

Many distributed data systems rely on a separate coordination service such as ZooKeeper to keep track of this cluster metadata. Each node registers itself in ZooKeeper, and ZooKeeper maintains the authoritative mapping of partitions to nodes. The routing tier or the partitioning-aware client can subscribe to this information in ZooKeeper.

A transaction is a way for an application to group several reads and writes together into a logical unit. Conceptually, all the reads and writes in a transaction are executed as one operation: either the entire transaction succeeds (commit) or it fails (abort, rollback). They are ultimately an abstraction layer that allow an application to pretend that certain concurrency problems and certain kinds of hardware and software faults don’t exist.

ACID

The safety guarantees provided by transactions are often described by the well known acronym ACID, which stands for Atomicity, Consistency, Isolation, and Durability.

Isolation

Isolation tries to hide concurrency issues from application developers. However, it does not operate as a binary (on/off). There are many different levels of isolation, and they all protect against different race conditions and concurrency issues, with varying tradeoffs. Before we talk about isolation levels, let’s understand what concurrency issues can arise.

General Concurrency Problems

Phantom Concurrency Problems

When a write in one transaction changes the result of a search query in another transaction, it is called a phantom. The problem with phantoms is that there is no object to which we can attach locks. To solve this, we can employ an approach called materializing conflicts. However, doing this is ugly, and should be considered a last resort.

Pessimistic & Optimistic Concurrency Control

Database Locking Mechanisms

Isolation Levels

Serializable Isolation Implementations

Async vs Sync Network

TCP vs UDP

Detecting Faults

A timeout is the only way of detecting a fault. But then how long should the timeout be?

• A long timeout means a long wait until a node is declared dead
• A short timeout detects faults faster, but carries a higher risk of incorrectly declaring a node dead during a temporary slowdown

A good timeout value can only be determined experimentally.

Clocks

It doesn’t make sense to think of a clock reading as a point in time. It is more like a range of times, within a confidence interval. Unfortunately, most systems don’t expose this uncertainty.

Timing

Unexpected Pauses

A node in a distributed system must assume that its execution can be paused for a significant length of time at any point. A thread or process can pause on a node for a number of reasons:

• The JVM’s garbage collector stops all running threads. Can be mitigated by treating this like a planned outage.
• In virtualized environments, a virtual machine can be suspended, which pauses the execution of all processes
• A user can close their laptop
• OS can context-switch to another thread
• Thread may be paused waiting on disk I/O
• A UNIX process receives a SIGSTOP signal

During the pause, the rest of the nodes keep operating and may even declare the node dead. Eventually, the paused node may continue running without even noticing that it was asleep until it checks its clock sometime later.

For some systems, this is unacceptable. These are known as real-time systems. In these systems, there is a specified deadline by which the software must respond. They are most commonly used in safety-critical embedded devices (i.e. airbags). They are very difficult and expensive to develop. For most server-side data processing systems, real-time guarantees are simply not economical or appropriate.

Lease

You can think of a lease as a lock with a timeout. It’s useful in a distributed system in lots of cases, such as determining who is the leader. Only one node can hold the lease at any one time. Thus, when a node obtains a lease, it knows it’s the leader until the lease expires. In order to remain a leader, the node must periodically renew the lease before it expires. If the node fails, it stops renewing the lease, so another node can take over when it expires.

However, there’s an issue: what if…

1. Node obtains the lease
2. Node pauses
3. Lease expires
4. Node unpauses

In this case, the node will believe it is still the leader. This can be resolved using fencing.

When a lease or lock is administered, a fencing token is also given. This number increases every time a lock or lease is granted. A server will only process requests from the fencing token with the highest number. So if the server processed a fencing token of 34, and then someone comes along with 33, it’ll register that fencing token as outdated, and not process the request. This, of course, opens you up to the vulnerability of dishonest nodes.

Byzantine

Distributed Algorithm Properties

Consistency Models

Other consistency models include:

The relationship between all these consistency models is visualized as follows:

When Do You Want Linearizability?

• Uniqueness constraints as data is written, such as usernames, emails, file names, etc.
• Bank account balance never goes negative
• You don’t sell more items than you have in stock
• Don’t double-book same seat on flight

Serializability

CAP Theorem

CAP theorem is similar to ACID, in that it is vague and unhelpful. It is sometimes presented as: Consistency, Availability, Partition Tolerance: pick 2 out of the 3. However, this is misleading because network faults always happen eventually. So, the real question is, when a network fault does inevitably happen, you can either choose to be consistent or available.

How Do You Achieve Linearizability?

Lamport Timestamps

Lamport timestamps are sequence numbers that are consistent with causality and provide a total order. They take on the form of (counter, nodeID). If you are given two lamport timestamps, the one with a greater counter value is greater. If the counter values are equal, the one with the greater node ID is greater.

Every node and every client keeps track of the maximum counter value it has seen so far, and includes that maximum on every request. When a node receives a request or response with a maximum counter value greater than its own counter value, it immediately increases its own counter to that maximum.

Lamport timestamps are not enough to achieve linearizability though. Total order of operation only emerges after you have collected all of the operations. If one operation needs to decide right now whether a decision should be made (e.g. 2 concurrent writes for a uniqueness constraint), it might need to check with every other node that there’s no concurrently executing operation that could affect its decision.

Total Order Broadcast

In order to use total ordering between multiple nodes, we should use total order broadcast / atomic broadcast, which is a message exchange protocol that guarantees that no messages are lost, and every message is delivered to every node in the same order and exactly once. Total order broadcast is used in database replication, serializable transactions, creating messages log, and lock for fencing tokens.

Total order broadcast is based on a principle known as state machine replication: If every message represents a write to the database, and every replica processes the same writes in the same order, then the replicas will remain consistent with each other.

Consensus

The goal of consensus is to get several nodes to agree on something. This is formalized as: one or more nodes may propose values, and the consensus algorithm decides on one of those values. A consensus algorithm must satisfy the following properties:

Most consensus algorithms (Paxos, Raft, Zab, etc.) are actually total order broadcast algorithms, because the nodes decide on a sequence of values, rather than a single value. Total order broadcast itself is equivalent to repeated rounds of consensus.

Most consensus algorithms also use a leader, but don’t guarantee that the leader is unique. They instead define epoch numbers, and guarantee that for each epoch, the leader is unique. Every time the current leader is thought to have died, an election begins with an incremented epoch number. If two different nodes think they’re both the leader, they compare epoch numbers. They know the one with the higher epoch number is the true leader.

Limitations of Consensus

Problems Consensus Solves

Two-Phase Commit (2PC)

2PC is a consensus algorithm (except that it doesn’t meet the termination property) that aims to achieve atomic transaction commits across multiple nodes. This is easy enough to do on a single node, and is usually handled by the storage engine in that case.

So how does 2PC address this?

1. The application reads and writes data on multiple database nodes as normal
2. When it’s ready to commit, the coordinator sends a “prepare” request to each of the nodes, asking them if they’re able to commit
3. The point at which the coordinator has received all responses and makes a decision is known as the commit point. At this point, the coordinator writes its commit or abort decision down into a transaction log.
4. If any of the nodes respond “no”, an “abort” request is sent to all nodes
5. Otherwise, the coordinator sends a “commit” request and the commit actually takes place

The entirety of 2PC hinges on the success of the transaction coordinator. If the coordinator has crashed and takes 20 minutes to start up again, locks will also be held by pending commits for 20 minutes. If the coordinator’s log is entirely lost for some reason, those locks will be held forever (or at least until the situation is manually resolved).

The transaction coordinator is itself a kind of database (in which transaction outcomes are stored), so it needs to be approached like one. That is why it’s typically replicated. We don’t want it to be a single point of failure.

Distributed Transactions

(Y’know, as opposed to single node transactions)

X/Open XA (short for eXtended architecture) is a standard for implementing 2PC across heterogeneous technologies. It is a C API for interfacing with a transaction coordinator. XA is supported by many traditional relational databases, as well as message brokers. It allows you to atomically commit transactions between different technologies.

However, it cannot detect deadlocks across different systems, and it does not work with Serializable Snapshot Isolation.

Membership and Coordination Services

ZooKeeper is the most common membership and coordination service. Others include etcd and Consul. It is essentially a key-value store that is designed to hold small amounts of data that can fit entirely in memory. This data is replicated across all nodes in a system using a fault-tolerant total order broadcast algorithm (consensus). Lots of tools rely on ZooKeeper specifically (HBase, Hadoop, Kafka, etc.). Along with consensus, it comes with the following features:

• Linearizable atomic operations
• Total ordering of operations
  • Provides a fencing token via a transaction ID and a version number
• Failure detection
  • Uses heartbeats
• Change notifications
  • A client can find out when another client joins the cluster, or fails
• Service discovery
  • Find out which IP addresses you need to connect to in order to reach a particular service

If you need a feature that is reducible to consensus, it is best to use ZooKeeper.

Batching processing systems take a large amount of input data, run a job to process it, and produce some output data. They are often scheduled periodically.

MapReduce

MapReduce is a batch processing algorithm. It lets you write code to process large datasets in a distributed file system, and thus, works in parallel across multiple machines easily.

It is implemented in various open source data systems, such as Hadoop, CouchDB, and MongoDB. Other frameworks have functionality built atop Hadoop’s MapReduce API, which is somewhat cumbersome. Some of these frameworks are: Pig, Hive, Cascading, and Crunch. Hadoop has the most commonly referenced implementation of MapReduce, so we’ll focus on that.

MapReduce treats inputs as immutable, and avoids side effects (such as writing to external databases). This allows for good performance and maintainability. It is designed to tolerate frequent unexpected task termination.

MapReduce is based on two eponymous functions: the mapper function and the reducer function. They work in the following order:

1. A set of input files is read
2. The mapper function extracts a (key, value) pair from each input record. This makes it suitable for sorting.
3. Behind the scenes, the shuffle happens. This means that the data is partitioned by reducer, sorted by key, and data partitions are copied from mappers to reducers.
4. The reducer function iterates over the sorted (key, value) pairs, and derives some information from them

Joins

How are joins performed in MapReduce? There are no indexes in MapReduce, and thus, no foreign keys. It must read the entire content of a file. There are two kinds of joins: map-side joins and reduce-side joins. They are categorized based on where the join logic happens. The variations are outlined below.

Workflows

MapReduce jobs can be chained together into workflows. The output of one job becomes the input to the next job. This works by having the first job’s output in HDFS be the second job’s input. If you have multiple jobs chained together, you might not care about the output of a single job. In that case, it’s just a means to an end, or intermediate state. The process of writing out this intermediate state is known as materialization, and ensures durability. While this means that if a workflow is terminated when it’s only 50% done and can then pick right back up where it left off, it has downsides:

Dataflow Engines

As an answer to the problems with MapReduce, other dataflow engines for distributed batch computations were developed, such as Spark, Tez, and Flink.

Rather than explicit map and reduce functions, these dataflow engines employ operators. The advantages they have over mapper and reducer functions are numerous:

However, it’s easy for MapReduce to recover from faults since it materializes intermediate state. Other dataflow engines take a different approach to tolerating faults. They try and recompute from other data that is still available.

Graphs

Let’s examine graph algorithms in the context of batch processing algorithms. These graph algorithms are often iterative, traversing one edge at a time and joining together vertices as they go. This is done until there are no more edges to traverse, or some other condition is met.

Recall that MapReduce must read the entire content of a file. Thus, performing this kind of graph algorithm with MapReduce is extremely inefficient. It will read the entire input dataset and produce a complete new output dataset, even if only a small part of the graph has changed compared to the last iteration.

An optimization for batch processing graphs was developed, known as the Bulk Synchronous Parallel (BSP) model of computation, or the Pregel model. In each iteration, a function is called for each vertex, passing it all the messages that were sent to it, much like a call to the reducer. Unlike MapReduce, the Pregel model remembers its state in memory from one iterate to the next. Various dataflow engines implement the Pregel Model, such as Spark and Flink.

Hadoop vs MPP Databases

Massively Parallel Processing (MPP) databases are a type of data warehouse where multiple nodes process analytic SQL queries in parallel. They are monolithic, tightly integrated pieces of software that take care of storage layout on disk, query planning, scheduling, and execution.

Stream processing software includes RabbitMQ, ActiveMQ, HornetQ, Qpid, TIBCO Enterprise Message Service, IBM MQ, Azure Service Bus, and Google Cloud Pub / Sub.

An event is a small, self-contained, immutable object containing the details of something that happened at some point in time. It is the data unit that comprises a stream, which is just any sequence of data that is made available over time. Related events are usually grouped together into a topic or stream. Events are created by producers / publishers / senders and processed by potentially multiple consumers / subscribers / recipients.

Brokers

A message broker / message queue acts as the middle man between these two groups. It is essentially a kind of database that is optimized for handling message streams.

Brokerless

There exists brokerless stream processing software, such as ZeroMQ and nanomsg. Alternatively, you can avoid using any of these kinds of software entirely and rely on UDP multicast. You can also use webhooks, which are a pattern in which a callback URL of one services is registered with another service, and it makes a request to that URL whenever an event occurs.

Methods For Turning Changes To A Database Into A Stream