Part 9. How to Store Data

A relational database stores data in tables, which represent a collection of related items, where each item is stored in a row, and each row in a table has the same columns. For example, if you were working on a website for a bank, and you needed to store data about the customers, you might have a customers table where each row represents one customer as a tuple of id, name, date_of_birth, and balance, as shown in Table 19.

Relational databases require you to define a schema to describe the structure of each table before you can write any data to that table. You’ll see how to define the schema for the customers table in Table 19 a little later in this blog post. For now, let’s imagine the schema already exists, and focus on how to read and write data. To interact with a relational database, you use a language called Structured Query Language (SQL). SQL is ubiquitous in the world of software, so it’s worth taking the time to learn it, as it will help you build applications, do performance tuning, perform data analysis, and more; I wish I had known it back when I was a summer intern!

The SQL to write data is an INSERT INTO statement, followed by the name of the table, the columns to insert, and the values to put into those columns. Example 141 shows how to insert the three rows from Table 19 into the customers table:

How do you know if these INSERT statements worked? One way is to try reading the data back out. To read data with a relational database, you use the same language, SQL, to formulate queries. The SQL syntax for queries is a SELECT statement, followed by the columns you wish to select, or the wildcard * for all columns, then FROM, followed by the name of the table to query. Example 142 shows how to retrieve all the data from the customers table:

As you’d expect, this query returns the three rows inserted in Example 141. You can filter the results by adding a WHERE clause with conditions to match. Example 143 shows a SQL query that selects customers born after 1950, which should return just two of the three rows:

SQL is an exceptionally flexible query language, but going into all the details is beyond the scope of this book (see [recommended_reading] to go deeper). All I’ll say for now is that one of the major advantages of the relational model is that it allows you to query your data in countless different ways: e.g., you can use WHERE to filter data, ORDER BY to sort data, GROUP BY to group data, JOIN to query data from multiple tables, COUNT, SUM, AVG, and a variety of other aggregate functions to perform calculations on your data, indices to make queries faster, and much more. If I had used a relational database for that load generator app when I was a summer intern, I could’ve replaced thousands of lines of custom query code with a dozen lines of SQL.

The flexibility and expressiveness of SQL is one of the many reasons most companies use relational databases as their primary data stores. Another major reason is due to ACID transactions, as discussed in the next section.

A transaction is a set of coherent operations that should be performed as a unit. In relational databases, transactions must meet the following four properties:

These four properties taken together form the acronym ACID, and it’s one of the defining properties of just about all relational databases. For example, going back to the bank example with the customers table, imagine that the bank charged a $100 annual fee for each customer. When the fee was due, you could use a SQL UPDATE statement to deduct $100 from every customer, as shown in Example 144:

A relational database will apply this change to all customers in a single ACID transaction: that is, either the transaction will complete successfully, and all customers will end up with $100 less, or no customers will be affected at all. This may seem obvious, but many of the data stores you’ll see later in this blog post do not support ACID transactions, so it would be possible for those data stores to crash part way through this transaction, and end up with some customers with $100 less and some unaffected.

Relational databases also support transactions across multiple statements. The canonical example is transferring money, such as moving $100 from the customer with ID 1 (Brian Kim) to the customer with ID 2 (Karen Johnson), as shown in Example 145:

All the statements between START TRANSACTION and COMMIT will execute as a single ACID transaction, ensuring that one account has the balance decreased by $100, and the other increased by $100, or neither account will be affected at all. If you were using one of the data stores from later in this blog post that don’t support ACID transactions, you could end up in an in-between state that is inconsistent: e.g., the first statement completes, subtracting $100, but then the data store crashes before the second statement runs, and as a result, the $100 simply vanishes into thin air. With a relational database, this sort of thing is not possible, regardless of crashes or concurrency. This is a major reason relational databases are a great choice as your company’s source of truth—something I wish I knew when building my load generator app as a summer intern! Another major reason is the support for schemas and constraints, as discussed in the next section.

Relational databases require you to define a schema for each table before you can read and write data to that table. To define a schema, you again use SQL, this time with a CREATE TABLE statement, followed by the name of the table, and a list of the columns. Example 146 shows the SQL to create the customers table in Table 19:

The preceding code creates a table called customers with columns called id, name, date_of_birth, and balance. Note that the schema also includes a number of integrity constraints to enforce business rules, such as the following:

Foreign key constraints are one of the defining characteristics of relational databases, as they allow you to define and enforce relationships between tables—this is what the "relational" in "relational database" refers to. This is critical in maintaining the referential integrity of your data, and it’s another major reason to use a relational database as your primary source of truth.

In addition to using CREATE TABLE to define the schema for new tables, you can use ALTER TABLE to modify the schema for existing tables (e.g., to add a new column). Carefully defining and modifying a schema is what allows you to evolve your data storage over time without running into backward compatibility issues, like I did with my load generator app.

Initially, you might manage schemas manually, connecting directly to the database and executing CREATE TABLE and ALTER TABLE commands by hand. However, as is often the case with manual work, this becomes error-prone and tedious. Over time, the number of CREATE TABLE and ALTER TABLE commands piles up, and as you add more and more environments where the database schema must be set up (e.g., dev, stage, prod), you’ll need a more systematic way to manage your database schemas. The solution, as you saw in Part 2, is to manage your schemas as code.

In particular, there are a number of schema migration tools that can help, such as Flyway, Liquibase, Atlas, Bytebase, Alembic, migrate, Squitch, ActiveRecord, Sequel, Knex.js, and GORM. These tools allow you to define your initial schemas and all the subsequent modifications as code, typically in an ordered series of migration files that you check into version control. For example, Flyway uses standard SQL in .sql files (e.g., v1_create_customers.sql, v2_create_accounts.sql, v3_update_customers.sql, etc.), whereas Knex.js uses a JavaScript DSL in .js files (e.g., 20240825_create_customers.js, 20240827_create_accounts.js, 20240905_update_customers.js, etc). You apply these migration files using the schema migration tool, which keeps track of which of your migration files have already been applied and which haven’t, so no matter what state your database is in, or how many times you run the migration tool, you can be confident your database will end up with the desired schema.

As you make changes to your app, new versions of the app code will rely on new versions of your database schema. To ensure these versions are automatically deployed to each environment, you will need to integrate the schema migration tool into your CI / CD pipeline (something you learned about in Part 5). One approach is to run the schema migrations as part of your app’s boot code, just before the app starts listening for requests. The main advantage of this option is that it works not only in shared environments (e.g., dev, stage, prod), but also in every developer’s local environment, which is not only convenient, but also ensures your schema migrations are constantly being tested. The main disadvantage is that migrations sometimes take a long time, and if an app takes too long to boot, some orchestration tools will think there’s a problem, and try to redeploy the app before the migration can finish. Also, if you are running serverless apps, which already struggle with cold starts, you shouldn’t add anything to the boot code that makes it worse. In these cases, you’re better off with a different approach, which is to run migrations as a separate step in your deployment pipeline, just before you deploy the app.

Now that you’ve seen the concepts behind relational databases, let’s see those concepts in action with a real-world example.

In this section, you’ll go through an example of deploying PostgreSQL, a popular open source relational database, in AWS using Amazon’s Relational Database Service (RDS), a fully-managed service that provides a secure, reliable, and scalable way to run several different types of relational databases, including PostgreSQL, MySQL, MS SQL Server, and Oracle Database. You’ll then manage the schema for this database using Knex.js and deploy a Lambda function and API Gateway to run a Node.js app that connects to the PostgreSQL database over TLS and runs queries.

Here are the steps to set this up:

Let’s start with creating the OpenTofu module.

Key-value stores are optimized for a single use case: fast lookup by a known identifier (key). They are effectively a hash table that is distributed across multiple servers. The idea is to deploy the key-value store between your app servers and your primary data store, as shown in Figure 97, so requests that are in the cache (a cache hit) are returned quickly, without having to talk to the primary data store, and only requests that aren’t in the cache (a cache miss) go to the primary data store (after which they are added to the cache to speed up future requests).

Some of the major players in the key-value store space include Redis / Valkey (Valkey is a fork of Redis that was created after Redis switched from an open source license to dual-licensing), Memcached, Amazon DynamoDB, and Riak KV. The API for most key-value stores primarily consists of just two types of functions, one to insert a key-value pair and one to look up a value by key. For example, with Redis, you use SET to insert a key-value pair and GET to look up a key:

Key-value stores do not require you to define a schema ahead of time (sometimes referred to as schemaless, but this is a misnomer, as you’ll learn later in this blog post), so you can store any kind of value you want. Typically, the values are either simple scalars (e.g., strings, integers) or blobs that contain arbitrary data that is opaque to the key-value store. Since the data store is only aware of keys and very basic types of values, other than operations by primary key, functionality is typically limited. That is, you shouldn’t expect support for flexible queries, joins, foreign key constraints, ACID transactions, or many of the other powerful features of a relational database.

You can deploy key-value stores yourself, or you can use managed services, such as Redis Cloud, Amazon ElastiCache, Google Cloud Memorystore, Azure Cache for Redis, and Upstash. Once you have a key-value store deployed, many libraries can automatically use them for cache-aside and write-through caching without you having to implement those strategies manually: e.g., WordPress has plugins that automatically integrate with Redis and Memcached and the Redis Smart Cache plugin provides automatic caching for any database you access via the Java Database Connectivity (JDBC) API.

Let’s now look at the second type of data store commonly used for caching, CDNs.

A content delivery network (CDN) consists of servers that are distributed all over the world, called Points of Presence (PoPs), that cache data from your origin servers (i.e., your app servers), and serve that data from a PoP that is as close to the user as possible. Whereas a key-value store goes between your app servers and your database, a CDN goes between your users and your app servers, as shown in Figure 98. When a user makes a request, it first goes to the PoP that is closest to that user, and if the content is already cached, the user gets a response immediately. If the content isn’t already cached, the PoP forwards the request to your origin servers, caches the response (to make future requests fast), and then returns it to the user.

Some of the major players in the CDN space include CloudFlare, Akamai, Fastly, Imperva, Amazon CloudFront, Google Cloud and Media CDN, and Azure CDN. CDNs offer several advantages:

CDNs are most valuable for content that (a) is the same for all of your users and (b) doesn’t change often. For example, news publications can usually offload a huge portion of their traffic to CDNs, as once an article is published, every user sees the same content, and that content isn’t updated too often. On the other hand, social networks and collaborative software can’t leverage CDNs as much, as every user sees different content, and the content changes often.

One place where virtually all companies can benefit from a CDN is when serving completely static content, such as images, videos, binaries, JavaScript, and CSS. Instead of having your app servers waste CPU and memory on serving up static content, you can offload most of this work to a CDN. In fact, many companies choose not to have their app servers involved in static content at all, not even as an origin server for a CDN, and instead offload all static content to dedicated file servers and object stores, as described in the next section.

A file server is a server that is designed to store and serve static content, such as images, videos, binaries, JavaScript, and CSS, so that your app servers can focus entirely on serving dynamic content (i.e., content that is different for each user and request). Requests first go to a CDN, which returns a response immediately if it is already cached, and if not, the CDN uses your app servers and file servers as origin servers for dynamic and static content, respectively, as shown in Figure 99:

Most web server software can easily be configured to serve files: e.g., Apache, Nginx, HAProxy Varnish, Lighttpd, Caddy, and Microsoft IIS. The hard part is handling storage, metadata, security, and scalability and availability:

Solving these problems for a small number of files can be straightforward, but things get trickier as the number of files increases. By the time you’re at a scale of a Snapchat, where users upload more than 4 billion pictures per day, storage, security, metadata, and scalability and availability are considerable challenges that require lots of custom tooling, huge numbers of servers and hard drives, RAID, NFS, and so on.

One way to make these challenges easier is to offload much of this work to an object store, as discussed in the next section.

An object store (sometimes called a blob store) is a system designed to store opaque objects or blobs, often in the form of files with associated metadata. Typically, these are cloud services, so you can think of object stores as a file server as a service. The major players in this space are Amazon Simple Storage Service (S3) (which you used in Part 5 to store OpenTofu state files), Google Cloud Storage (GCS), Azure Blog Storage, CloudFlare R2, Wasabi, and Backblaze. Object stores provide out-of-the-box solutions to the challenges you saw with file servers in the previous section:

Many object stores also provide a variety of other useful features, such as automatic archival or deletion of older files (to save money), replication across data centers in different regions, search and analytics across all the files you store in the object store (e.g., Amazon Athena allows you to use SQL to query CSV, JSON, ORC, Avro, or Parquet files stored in S3), integration with compute to help automate workflows (e.g., you can have S3 automatically trigger a Lambda function each time you upload a file), and more. This combination of features is why even companies who otherwise keep everything on-prem often turn to the cloud and object stores for file storage.

To get a better sense for file storage, let’s go through an example.

In this section, you’ll deploy an S3 bucket, and configure it to host a static website, and then deploy CloudFront in front of this website as a CDN. This is a good model to follow if you want low-effort, scalable, highly available, globally-distributed static content hosting. To set this up, you’ll need to go through the following three steps:

Let’s start with creating the S3 bucket.

To get a sense of how document stores work, let’s use MongoDB as an example. MongoDB allows you to store JSON documents in collections, somewhat analogously to how a relational database allows you to store rows in tables. MongoDB does not require you to define a schema for your documents, so you can store JSON data in any format you want. To read and write data, you use the MongoDB Query Language (MQL), which is similar to JavaScript. Example 164 shows how you can use the insertOne command to store a JSON document in a collection called bank:

This is the same bank example you saw with relational databases earlier in this blog post. Example 165 shows how you can use the insertMany command to insert two more JSON documents so that the bank collection has the same data as Table 19:

To read data back out, you can use the find command as shown in Example 166:

You get back the exact documents you inserted, except for one new item: MongoDB automatically adds an _id field to every document, which it uses as a unique identifier, and a key for lookups, similar to a key-value store. For example, you can look up a document by ID as shown in Example 167:

The big difference between key-value stores and document stores is that document stores can natively understand and process the full contents of each document, rather than treating them as opaque blobs. This gives you richer query functionality. Example 168 shows how you can find all customers born after 1950, the same query you saw in SQL in Example 143:

You also get richer functionality when updating documents. Example 169 shows how you can use the updateMany command to deduct $100 from all customers, similar to the SQL UPDATE you saw in Example 144:

The first argument to updateMany is a filter to narrow down which documents to update; the preceding code sets this to an empty object, so the command will update all documents. The second argument is the update operation to perform; the preceding code uses the $inc operator to increment all balances by -100.

All of this richer querying and update functionality is great, but there are two major limitations. First, most document stores do not support working with multiple collections: that is, there is no support for joins.^[34] Second, most document stores don’t support ACID transactions, as discussed in the next section.

There is a serious problem with the code in Example 169: most document stores don’t support ACID transactions.^[35] You might get atomic operations on a single document (e.g., if you updated one document with the updateOne command), but you rarely get it for updates to multiple documents. That means it’s possible for that code to deduct $100 from some customers but not others: e.g., if MongoDB crashes in the middle of the updateMany operation.

This is not at all obvious from the code, and many developers who are not aware of this are caught off guard when their document store operations don’t produce the results they expect. This is one of many gotchas with using non-relational databases, especially as your source of truth. Other major gotchas include dealing with eventual consistency, as you’ll see later in this blog post, and the lack of support for schemas and constraints, as discussed in the next section.

Most document stores do not require you to define a schema or constraints up front. This is sometimes referred to as schemaless, but that’s a bit of a misnomer. The reality is that there is always a schema. The only question is whether you enforce a schema-on-read or a schema-on-write. Relational databases enforce a schema-on-write, which means the schema and constraints must be defined ahead of time, and the database will only allow you to write data that matches the schema and constraints. Most document stores, such as MongoDB, don’t require you to define the schema or constraints ahead of time, so you can structure your data however you want, but eventually, something will read that data, and that code will have to enforce a schema-on-read to be able to parse the data and do something useful with it. For example, to parse data from the bank collection you saw in the previous section, you might create the Java code shown in Example 170:

The Java class in Example 170 defines a schema and constraints: i.e., you’re expecting field names such as name and balance with types String and int, respectively. More accurately, this is an example of schema-on-read, as this class defines the schema you’re expecting from the data store, and either the data you read matches the Customer data structure, or you will get an error. Since document stores don’t enforce schemas or constraints, you can insert any data you want in any collection, such as the example shown in Example 171:

Did you catch the error? The preceding code uses birth_date instead of date_of_birth. Whoops. MongoDB will allow you to insert this data without any complaints, but when you try to parse this data with the Customer class, you may get an error. And this is just one of many types of errors you may get with schema-on-read. Since most document stores don’t support domain constraints or foreign key constraints, you will also have to worry about typos in field names, null or empty values for required fields, incorrect types for fields, IDs that reference non-existent documents in other collections, and so on.

Dealing with these errors when you read the data is hard, so it’s better to prevent these errors in the first place by blocking invalid data on write. That’s an area where schema-on-write has a decided advantage, as it allows you to ensure your data is well-formed by enforcing a schema and constraints in one place, the (well-tested) data store, instead of trying to enforce it in dozens of places, including in every part of your application code, every script, every console interaction, and so on.

That said, schema-on-read is advantageous if you are dealing with semi-structured or non-uniform data. I wouldn’t use a document store for highly structured bank data, but I might use one for user-generated documents, event-tracking data, and log messages. Schema-on-read can also be advantageous if the schema changes often. With a relational database, certain types of schema changes take a long time or even require downtime. With schema-on-read, all you have to do is update your application code to be able to handle both the new data format and the old one, and your migration is done. Or, to be more accurate, your migration has just started, and it will happen incrementally as new data gets written.

There’s one other trade-off to consider between schema-on-read and schema-on-write: performance. With schema-on-write, as with a relational database, the data store knows the schema ahead of time, and the schema is the same for all the data in a single table, so the data can be stored very efficiently, both in terms of disk space usage, and the performance of disk lookup operations. With schema-on-read, as with a document store, since each document can have a different schema, the data store has to store the schema with that document, which is less efficient. This is one of the reasons that data stores that are designed for performance and efficiency typically use schema-on-write. This includes data stores designed to extract insights from your data using analytics, as discussed in the next section.

On the surface, columnar databases (AKA column-oriented databases) look similar to relational databases, as they store data in tables that consist of rows and columns, they usually have you define a schema ahead of time, and sometimes, they support a query language that looks similar to SQL. However, there are a few major differences. First, most columnar databases do not support ACID transactions, joins, foreign key constraints, and many other key relational database features. Second, the key design principle of columnar databases, and the source of their name, is that they are column-oriented, which means they are optimized for operations across columns, whereas relational databases are typically row-oriented, which means they are optimized for operations across rows of data.

This is best explained with an example. Consider the books table shown in Table 20:

How does this data get stored on the hard drive? In a row-oriented relational database, the values in each row will be kept together, so conceptually, the serialized data might look similar to what you see in Example 172:

Compare this to the way a column-oriented store might serialize the same data, as shown in Example 173:

In this format, all the values in a single column are laid out sequentially, with the column values as keys (e.g., 1993), and the IDs as values (e.g., 2,3). Now consider the query shown in Example 174:

Because this query uses SELECT *, it will need to read every column for any matching rows. With the row-oriented storage in Example 172, the data for all the columns in a row is laid out sequentially on the hard drive, whereas with the column-oriented storage in Example 173, the data for each column is scattered across the hard drive. Hard drives perform better for sequential reads than random reads, so for this sort of query, especially with a large amount of data (think millions or billions of rows), the row-oriented approach will be considerably faster. Compare that to the query in Example 175:

This query uses an aggregate, SELECT COUNT(*), so it will only need to read the values in the year_published column to satisfy the WHERE clause, and then count how many matches there are. With the row-oriented storage in Example 172, this requires jumping all over the hard drive to read the year_published value for each row, whereas with the column-oriented storage in Example 173, all the data for the year_published column is laid out sequentially, so for this sort of query, with millions or billions of rows, the column-oriented approach will be considerably faster. When you’re doing analytics, aggregate functions such as COUNT, SUM, AVG come up all the time, which is why the column-oriented approach is used in a large number of analytics use cases, as described in the next section.

The analytics space is massive, and a detailed overview of all the tools is beyond the scope of this book. In this section, I’ll only call out a few of the most common categories of tools you’re likely to run across. Note that these categories are not mutually exclusive, so some data stores may fit into more than one of these categories.

The diagram in Figure 101 is highly simplified. First, what look like simple arrows from the various systems to the data warehouse are actually complicated background processes known as extract, transform, and load (ETL), where you use specialized software, such as Apache Airflow, Oracle Data Integrator, SQL Server Integration Services, AWS Glue, Azure Data Factory, Google Cloud Dataflow, Stitch, Qlik, Informatica, Matillion, and Integrate.io, to extract data from one system that uses one format, transform it into the format used by another system (cleaning up and standardizing the data along the way), and then load it into that other system. Second, there are not only arrows from each system to the data warehouse, but arrows between many of the other systems, too, representing background jobs, event-based communication, and so on. All of this falls into the realm of asynchronous processing, which is the topic of the next section.

A message queue is a data store that can be used for asynchronous communication between producers, who write messages to the queue, and consumers, who read messages from the queue, as shown in Figure 102. Some of the most popular message queues include RabbitMQ, ActiveMQ, ZeroMQ, Amazon Simple Queue Service (SQS), Google Cloud Tasks, and Azure Queue Storage.

The typical process of using a queue is:

Queues are most often used for tasks that run in the background, as opposed to tasks you do during a live request from a user. For example, if you are building an app that lets users upload images, if you need to process each image (e.g., create copies of the image in different sizes for web, mobile, thumbnail previews, etc.), you may want to do that in the background, rather than making the user wait for it. To do that, your frontend server stores the original image on a file server and adds a message to a queue with the location of the image. Later on, a separate consumer process reads the message from the queue, downloads the image from the file server, processes the image, and when it’s done, deletes the message from the queue. Other common use cases include encoding videos, sending email campaigns, delivering notifications, generating reports, and order processing.

Using queues for asynchronous communication between services provides several key benefits:

Whereas message queues are used for one-to-one communication between a producer and a consumer, event streams are used for one-to-many communication, as discussed in the next section.

Event streaming tools allow services to communicate asynchronously in a manner similar to a message queue, but the main difference is that instead of each message being consumed by a single consumer, streaming allows each message to be consumed by multiple consumers, as shown in Figure 103. Some of the most popular event streaming tools include Apache Kafka, Confluent, Amazon Managed Streaming for Kafka (MSK), Amazon Kinesis, Amazon EventBridge, Google Cloud Managed Service for Kafka, Google Cloud Pub/Sub, Azure HDInsight, Apache Pulsar, NATS, and Redpanda.

The typical process of using event streaming is:

At its most basic level, event streaming could be used as a replacement for a message queue to allow services to communicate asynchronously, but this is not the primary use case. Whereas a message queue is typically used to allow service A to send a message specifically destined for service B, the idea with event streaming is for every service to publish a stream of events that represent important data points or changes in state in that service, but aren’t necessarily designed for any one specific recipient, which allows for multiple other services to subscribe and react to whatever streams of events are relevant to them. This is known as an event-driven architecture.

The difference between messages in a message queue and events in an event stream has a profound impact on how you build your services. In Figure 101, you saw a simplified diagram showing all systems sending their data to a data warehouse. Figure 105 shows a slightly more realistic image:

As the number of services grows, the number of connections between them—whether those are synchronous or asynchronous connections (via a message queue)—grows even faster. With N services, you end up with roughly N² connections, across a variety of interfaces and protocols that often require complicated ETL. Setting up and maintaining all these connections can be a massive undertaking. Event streaming offers an alternative solution, as shown in Figure 106:

In Part 7, you saw that a network switch allows you to connect N computers with N cables (each computer has one cable connected to the switch) instead of N². Analogously, an event streaming platform allows you to connect N services with N connections (each service has one connection to the event streaming platform) instead of N². Dramatically simplified connectivity is one of the major benefits of an event-driven architecture. Another major benefit, and one that’s less obvious, is that an event-driven architecture allows you to add new services—new consumers—without having to modify any existing producers.

An example can help illustrate the power of this concept. First, consider an architecture where services message each other directly, whether synchronously or asynchronously via a message queue. For example, service A might send the message "a new image has been uploaded to location X, please process that image" to service B. 6 months later, you want to add a new service C to scan images for inappropriate content. In order for this service to do its job, you have to update service A to send an additional message "a new image has been uploaded to location X, please scan that image for inappropriate content" to service C.

Now compare this to an event-driven architecture, where service A doesn’t have to know about the existence of other services at all. Service A merely publishes important events, such as "a new image has been uploaded to location X." Perhaps on day one, service B subscribes to this event stream, and is able to process each image; 6 months later, when you add service C, it can subscribe to the same event stream to start scanning images for inappropriate content—without any need to modify service A. You could add dozens more services that consume service A’s event stream, again, with no need for A to be aware of them at all.

In an event-driven architecture, every service publishes important events, such as "a new user has registered," "a user clicked a button," "an order has been placed," "a server is down," and so on. Any other service can subscribe to any of these events streams to perform a variety of actions: e.g., update a search index, detect fraudulent activity, generate a report, send out a notification, and so on. Moreover, each time a service subscribes to an event stream, it can choose to start at offset 0 in that stream (refer to Figure 104), effectively "going back in time," and processing all the historical events from that event stream (e.g., all images that have ever been uploaded) until it catches up to the latest offset, or it can start immediately at the latest offset, and just process new events.

Event-driven architectures provide a large number of benefits:

Let’s now move on to the next section, which focuses not on a specific data storage use case, but on the general problem of data store scalability and availability.

Most relational databases are designed to run on a single server. However, as you learned in Part 3, a single server is a single point of failure—a bottleneck to scalability and availability. This isn’t something you have to worry about until you are storing tremendous amounts of data and serving a huge amount of traffic, but if you get to that point, you should be aware that it’s easy to scale a database vertically by making a single server more powerful (more CPU, more memory, more disk space), but it’s harder to scale a database horizontally across multiple servers. To horizontally scale a relational database—or any data store—there are two primary strategies, replication and partitioning, as discussed in the next two sections.

In the mid-to-late 2000s, the challenges with scalability and high availability for relational databases led to the creation of a number of non-relational databases, often called NoSQL databases. NoSQL, which at various times stood for Non-SQL or Not-Only-SQL, is a fuzzy term that refers to databases that do not use SQL or the relational model. Over the years, there have been many types of non-relational databases, most of which failed to gain wide adoption (e.g., object databases in the 90s, XML databases in the early 2000s), but NoSQL in particular refers to databases that were built in the late 2000s, primarily by Internet companies struggling to adapt relational databases to unprecedented demands in performance, availability, and data volume.

The early inspirations for NoSQL included Google’s 2006 paper on BigTable and Amazon’s 2007 paper on Dynamo. The actual term "NoSQL" came after these papers, originating as a Twitter hashtag (#NoSQL) for a 2009 meetup^[37] in San Francisco to discuss "open source, distributed, non-relational databases," which is still the best definition of NoSQL that we have. The primary types of data stores that fall under the NoSQL umbrella are key-value stores, document stores, and columnar databases, all of which you’ve already seen in this blog post.^[38]

Most NoSQL databases were designed from the ground up for scalability and availability, so the default deployment often includes replication and partitioning. For example, MongoDB is typically deployed in a cluster that consists of multiple shards, where each shard has a primary (for writes) and one or more replicas (for reads), plus dedicated servers that handle query routing, auto-sharding, and auto-rebalancing. The benefit is that you get a highly scalable and available data store. The cost is that these are complicated distributed systems, and that comes with a number of challenges, as you’ll see in the next section. For now, I’ll just say that, in the pursuit of scalability and high availability, most NoSQL data stores sacrifice key features from relational databases, such as ACID transactions, referential integrity, and a flexible query language (SQL) that supports joins.

For some use cases, this was too many sacrifices, which led to the creation of a new breed of relational database in the mid-to-late 2010s, often called NewSQL, that tried to retain the strengths of a relational database (e.g., ACID transactions, SQL), while providing better availability and scalability. Some of the major players in this space include Google Spanner, Amazon Aurora, CockroachDB, YugabyteDB, and VoltDB. Under the hood, these are also complex distributed systems that use replication and partitioning to achieve high scalability and availability, but they try to use new techniques to not sacrifice too many relational database benefits along the way. The approaches they use are fascinating, but beyond the scope of this blog post series, especially as many of the early NewSQL players died out, and the ones that remain are still relatively young and immature, so I won’t spend much time on them in this blog post.

Remember, data storage technology takes at least a decade to mature. As of the writing of this book, most NoSQL data stores are 10-15 years old, so they are just starting to become mature and reliable systems. Most NewSQL systems are still less than 10 years old, so they are still relatively young (at least as far as data storage technologies go). Given that both NoSQL and NewSQL databases are typically complex distributed systems, they face challenges that may take even more than a decade to solve, as discussed in the next section.

As you may remember from Section 6.2.3.11, distributed systems are complicated, and distributed data stores even more so. One of the challenges is that all distributed systems are subject to the CAP theorem, which gets its name from the following three properties:

All three of these are desirable properties for a distributed system, but the CAP theorem says you can only pick two. Moreover, in practice, no network can guarantee there will never be any partitions, so all real-world distributed systems have to provide partition tolerance—they have to pick P—which means you really only get to pick one more. That is, in the presence of a network partition, does your distributed system provide consistency (C) or availability (A)?

Some systems, such as HBase and Redis, pick C, so they try to keep data consistent on all nodes, and in the case of a network partition, they lose availability. If you use a data store that picks C, you have to accept that, from time to time, that data store will be down. Other systems, such as Cassandra, Riak, and CouchDB, pick A, so they are eventually consistent, which means that during a network partition, they will remain available, but different nodes may end up with different data. In fact, even without a partition, eventually consistent systems may have different data on different nodes, at least for a short amount of time. If you use a data store that picks A, then you have to deal with an eventually consistent data model, which can be confusing for programmers and users (e.g., you just updated some data, but after refreshing the page, you still see the old data). Some systems, such as MongoDB, allow you to tune for availability or consistency via configuration settings, allowing you to pick C or A depending on the use case.

Another challenge with distributed systems is that they introduce many new failure modes. At some point, every data store will fail. The question is, how many different ways can the system fail and how easy is it to understand and fix each one? Usually, the number and complexity of failure modes on a single-node system (e.g., a relational database) is far lower than on a distributed NoSQL or NewSQL system that has multiple writers, auto-sharding, auto-rebalancing, eventual consistency, consensus algorithms, and so on. For example, the complexity of the many different failure modes was one of the main reasons Pinterest had to move off Cassandra and Etsy had to move off MongoDB.

The third challenge is figuring out a business model to support these data systems. It takes a decade or two to build a reliable data store, and finding a way to sustainably pay developers during all that time is tricky. Many data store companies have shut down—e.g., RethinkDB, FoundationDB, GenieDB, ScaleDB, and many others—which is a huge problem if your company relies on these technologies for storing your most valuable asset! A data store that has been around 20+ years is not only more mature than a data store that just came out in the last few years, but it’s also more likely than the new data store to still be around another 20 years from now (this is called the Lindy effect).

This doesn’t mean you should avoid distributed systems, NoSQL, or NewSQL. It just means that you need to understand what they are good at, what they are not good at, and the risks you are taking on. For example, if you have extreme scale and availability requirements that you can’t handle with a relational database, and you have a team willing to put in the time and effort to deploy and maintain a NoSQL or NewSQL database, then by all means, go for it. But if you’re a tiny startup, with virtually no traffic, using a complex distributed data store right out of the gate might not be the right way to spend your limited resources.

Note that things can go wrong even with the most mature and battle-tested data store. Therefore, as the final topic of this blog post, let’s talk about how to manage backup and recovery to minimize the risk of losing data.

Below are the most common strategies for backing up data. Note that these strategies are not mutually exclusive. In fact, each strategy has different advantages and drawbacks, and protects against different types of disasters, so it’s usually a good idea to use several of these strategies to ensure you’re fully covered.

The following sections will go through each of these strategies, including their advantages and drawbacks, starting with scheduled disk backups.

There are a few practices I recommend when backing up your data:

Now that you’ve learned about backup strategies and recommended patterns, let’s put it into practice with a real-world example using PostgreSQL.

Earlier in this blog post, you created a lambda-rds root module that could deploy PostgreSQL in AWS using RDS. Let’s update that example to do the following:

Head over to the lambda-rds module, open up main.tf, find your usage of the rds-postgres module, and update it as shown in Example 176:

One of the benefits of using a managed service like RDS is that it makes it easy to enable common functionality, such as backups:

To add a read replica, add a second module block that uses the rds-postgres module, as shown in Example 177:

Again, using RDS makes it easy to use common functionality such as read replicas:

The read replica, as its name implies, is read-only, whereas the primary database accepts both reads and writes. To run schema migrations, you need both read and write access, so you should continue to use the primary URL for those. However, the Lambda function only needs read access for its one database query, so you can update it to talk to the read replica, rather than the primary, as shown in Example 178:

This code updates the environment variables passed to the Lambda function to use the replica hostname instead of the primary. Run apply to deploy these changes:

An RDS replica can take 5-15 minutes to deploy, so be patient. When apply completes, head over to the Lambda console, click on the "lambda-rds-app" function, select the Configuration tab, click on "Environment variables" on the left side, and you should see something similar to Figure 107:

The Lambda function should now have the DB_HOST set to the replica’s URL, and not the primary’s. If everything looks correct, test out the URL in the app_endpoint output variable one more time:

If you see the exact same results as before, congrats, that means your Lambda function is now reading data from a read replica! Moreover, your database now has backups in the form of daily snapshots.

When you’re done testing, commit your code, and run destroy to clean everything up. Note that, as part of the destroy process, RDS will take one final snapshot of the database, which is a handy failsafe in case you delete a database by accident.