The problem with most databases’ replication logs is that they have long been considered to be an internal implementation detail of the database, not a public API. Clients are supposed to query the database through its data model and query language, not parse the replication logs and try to extract data from them.
For decades, many databases simply did not have a documented way of getting the log of changes written to them. This made it difficult to take all the changes made in a database and replicate them to a different storage technology, such as a search index, cache, or data warehouse.
More recently, there has been growing interest in change data capture (CDC), which is the process of observing all data changes written to a database and extracting them in a form in which they can be replicated to other systems. CDC is especially interesting if changes are made available as a stream, immediately as they are written.
For example, you can capture the changes in a database and continually apply the same changes to a search index. If the log of changes is applied in the same order, you can expect the data in the search index to match the data in the database. The search index and any other derived data systems are just consumers of the change stream.
Figure 12-5 shows how the concurrency problem of Figure 12-4 is solved with CDC. Even though the two requests to set X to A and B, respectively, arrive concurrently at the database, the database decides on the order in which to execute them and writes them to its replication log in that order. The search index picks them up and applies them in the same order. If you need the data in another system, such as a data warehouse, you can simply add it as another consumer of the CDC event stream.
Diagram illustrating how changes made in a database by two clients are captured and logged in sequence using Change Data Capture (CDC), then applied in order to a search index and data warehouse.
Figure 12-5. Taking changes committed to a database and propagating them to downstream systems in the same order
Implementing CDC
We can call the log consumers derived data systems, as discussed in “Systems of Record and Derived Data”. The data stored in the search index and the data warehouse is just another view onto the data in the system of record. CDC is a mechanism for ensuring that all changes made to the system of record are also reflected in the derived data systems so that the derived systems have an accurate copy of the data.
Essentially, CDC makes one database the leader (the one from which the changes are captured) and turns the others into followers. A log-based message broker is well suited for transporting the change events from the source database to the derived systems, since it preserves the ordering of messages (avoiding the reordering issue of Figure 12-2).
Logical replication logs can be used to implement CDC (see “Logical (row-based) log replication”), although it comes with challenges, such as handling schema changes and properly modeling updates. The Debezium open source project addresses these challenges. The project contains source connectors for MySQL, PostgreSQL, Oracle, SQL Server, Db2, Cassandra, and many other databases. These connectors attach to database replication logs and surface the changes in a standard event schema. Messages can then be transformed and written to downstream databases. The Kafka Connect framework offers CDC connectors for various databases as well. Maxwell does something similar for MySQL by parsing the binlog, GoldenGate provides similar facilities for Oracle, and pgcapture does the same for PostgreSQL.
Like message brokers, CDC is usually asynchronous: the system-of-record database does not wait for a change to be applied to consumers before committing it. This design has the operational advantage that adding a slow consumer does not affect the system of record too much, but it has the downside that all the issues of replication lag apply (see “Problems with Replication Lag”).
Initial snapshot
If you have the log of all changes that were ever made to a database, you can reconstruct the entire state of the database by replaying the log. However, in many cases, keeping all changes forever would require too much disk space, and replaying it would take too long, so the log needs to be truncated.
Building a new full-text index, for example, requires a full copy of the entire database. Applying only a log of recent changes would not be sufficient, since it would be missing items that were not recently updated. Thus, if you don’t have the entire log history, you need to start with a consistent snapshot, as discussed in “Setting Up New Followers”.
The snapshot of the database must correspond to a known position or offset in the change log, so that you know at which point to start applying changes after the snapshot has been processed. Some CDC tools integrate this snapshot facility, while others leave it as a manual operation. Debezium uses Netflix’s DBLog watermarking algorithm to provide incremental snapshots.
Log compaction
If you can keep only a limited amount of log history, you need to go through the snapshot process every time you want to add a new derived data system. However, log compaction provides a good alternative.
We discussed log compaction previously in “Log-Structured Storage”, in the context of log-structured storage engines (see Figure 4-3 for an example). The principle is simple: the storage engine periodically looks for log records with the same key, throws away any duplicates, and keeps only the most recent update for each key. This can make log segments much smaller, so segments may also be merged as part of the compaction process, as shown in Figure 12-6. This process runs in the background.
Diagram illustrating log compaction and merging of two data file segments, retaining only the most recent values for each key, like video play counts.
Figure 12-6. In this log of key-value pairs, the key is the ID of a cat video (mew, purr, scratch, or yawn) and the value is the number of times it has been played; log compaction retains only the most recent value for each key.
In a log-structured storage engine, an update with a special null value (a tombstone) indicates that a key was deleted and causes it to be removed during log compaction. But as long as a key is not overwritten or deleted, it stays in the log forever. The disk space required for such a compacted log depends only on the current contents of the database, not the number of writes that have ever occurred in the database. If the same key is frequently overwritten, previous values will eventually be garbage-collected, and only the latest value will be retained.
The same idea works in the context of log-based message brokers and CDC. If the CDC system is set up such that every change has a primary key, and every update for a key replaces the previous value for that key, then it’s sufficient to keep just the most recent write for a particular key.
Now, whenever you want to rebuild a derived data system such as a search index, you can start a new consumer from offset 0 of the log-compacted topic and sequentially scan over all messages in the log. The log is guaranteed to contain the most recent value for every key in the database (and maybe some older values). In other words, you can use it to obtain a full copy of the database contents without having to take another snapshot of the CDC source database.
This log compaction feature is supported by Apache Kafka. As we shall see later in this topic, it allows the message broker to be used for durable storage, not just for transient messaging.
API support for change streams
Most popular databases now expose change streams as a first-class interface, rather than the retrofitted and reverse-engineered CDC efforts of the past. Relational databases such as MySQL and PostgreSQL typically send changes through the same replication log they use for their own replicas. Most cloud vendors offer CDC solutions for their products as well—for example, Datastream offers streaming data access for Google Cloud’s relational databases and data warehouses.
Even eventually consistent, quorum-based databases such as Cassandra now support CDC. As we saw in “Implementing Linearizable Systems”, clients must persist writes to a majority of nodes before they’re considered visible. CDC support for quorum writes is challenging because there’s no single source of truth to subscribe to. Whether the data is visible depends on each reader’s consistency preferences. Cassandra sidesteps this issue by exposing raw log segments for each node rather than providing a single stream of mutations. Systems that wish to consume the data must read the raw log segments for each node and decide how best to merge them into a single stream (much as a quorum reader does).
Kafka Connect integrates CDC tools for a wide range of database systems with Kafka. Once the stream of change events is in Kafka, it can be used to update derived data systems such as search indexes as well as feed into stream processing systems, as discussed later in this topic.
CDC versus event sourcing
How does CDC compare to event sourcing? Similarly to CDC, event sourcing involves storing all changes to the application state as a log of change events. The biggest difference is that event sourcing applies the idea at a different level of abstraction:
-
In CDC, the application uses the database in a mutable way, updating and deleting records at will. The log of changes is extracted from the database at a low level (e.g., by parsing the replication log), which ensures that the order of writes extracted from the database matches the order in which they were actually written, avoiding the race condition in Figure 12-4.
-
In event sourcing, the application logic is explicitly built on the basis of immutable events that are written to an event log. In this case, the event store is append-only, and updates or deletes of events are discouraged or prohibited. Events are designed to reflect things that happened at the application level rather than low-level state changes.
Which one is better depends on your situation. Adopting event sourcing is a big change for an application that is not already doing it; it has a number of pros and cons, which we discussed in “Event Sourcing and CQRS”. In contrast, CDC can be added to an existing database with minimal changes; the application writing to the database might not even know that CDC is occurring.
Change Data Capture and Database Schemas
Though CDC appears easier to adopt than event sourcing, it comes with its own set of challenges. In a microservices architecture, a database is typically accessed from only one service. Other services interact with it through that service’s public API, but they don’t normally access the database directly. This makes the database an internal implementation detail of the service, allowing the developers to change its schema without affecting the public API.
However, CDC systems typically use the upstream database’s schema when replicating its data, which turns these schemas into public APIs that must be managed much like the public API of the service. Removing a column in the database table will break downstream consumers that depend on that field. Such challenges have always existed with data pipelines, but they typically impacted only data warehouse ETL. Since CDC is often implemented as a data stream, other production services might be consumers. Breaking such consumers can cause a customer-facing outage. Data contracts are often used to prevent these breakages.
A common way to decouple internal from external schemas is to use the outbox pattern. Outboxes are tables with their own schemas, which are exposed to the CDC system rather than the internal domain model in the database. Developers can then modify their internal schemas as they see fit while leaving their outbox tables untouched. This might look like a dual write—it is. However, outboxes avoid the challenges we discussed in “Keeping Systems in Sync” by keeping both writes in the same system (the database). This design allows both writes to appear in a single transaction.
Outboxes present a few trade-offs, though. Developers must still maintain the transformation between their internal and outbox schemas, which can be challenging. An outbox also increases the amount of data that the database has to write to its underlying storage, which might trigger performance problems.
As with CDC, replaying the event log allows you to reconstruct the current state of the system. However, log compaction needs to be handled differently:
-
A CDC event for the update of a record typically contains the entire new version of the record, so the current value for a primary key is entirely determined by the most recent event for that primary key, and log compaction can discard previous events for the same key.
-
On the other hand, with event sourcing, events are modeled at a higher level. An event typically expresses the intent of a user action, not the mechanics of the state update that occurred as a result of the action. In this case, later events typically do not override prior events, so you need the full history of events to reconstruct the final state. Log compaction is not possible in the same way.
Applications that use event sourcing typically have a mechanism for storing snapshots of the current state that is derived from the log of events, so they don’t need to repeatedly reprocess the full log. However, this is only a performance optimization to speed up reads and recovery from crashes; the intention is that the system is able to store all raw events forever and reprocess the full event log whenever required. We discuss this assumption in “Limitations of immutability”.