How Change Data Capture (CDC) Works with Streaming Database
Tags: database, sql, webdev, opensource
Published: 2023-04-07 18:50:26

## Efficiently capture databases changes with a streaming database

Streaming [Change Data Capture](https://www.confluent.io/learn/change-data-capture/) (CDC) is a data integration approach that has become increasingly popular in recent years. Streaming CDC is a technique that enables organizations to capture and transmit real-time data changes in data sources like SQL/NoSQL databases and sent them to a target system, such as a data warehouse or analytics platform, where they can be used to generate insights and drive business decisions. Streaming CDC can be particularly useful for organizations that need to react quickly to changing market conditions or customer needs.

In this post, we will explore **how CDC works** with streaming databases and **the benefits this integration provides.** You will also learn how to analyze data captured from **MySQL** with **RisingWave**.

> [RisingWave](https://www.risingwave-labs.com/) is an open-source streaming database that has built-in fully-managed CDC source connectors for [MySQL](https://www.risingwave.dev/docs/current/ingest-from-mysql-cdc/) and [PostgreSQL](https://www.risingwave.dev/docs/current/ingest-from-postgres-cdc/) and it allows you to query real-time streams using SQL. You can get a materialized view that is always up-to-date.

## Learning objectives

By the end of this article, you will learn:

- What is Change Data Capture (CDC)?
- CDC use cases.
- Why use CDC with Streaming Database?
- How to Ingest CDC data from MySQL using RisingWave connector.

![How Change Data Capture (CDC) works with Streaming Database (1).png](https://static.apiseven.com/uploads/2023/04/07/lwKf9Xg3_How%20Change%20Data%20Capture%20%28CDC%29%20works%20with%20Streaming%20Database%20%281%29.png)

## What is Change Data Capture?

_Change Data Capture_ is a technique used to capture and propagate changes made to a database such as [MySQL](https://www.mysql.com/), [Microsoft SQL](https://learn.microsoft.com/en-us/sql/relational-databases/track-changes/about-change-data-capture-sql-server?view=sql-server-ver16), [Oracle](https://docs.oracle.com/en/database/oracle/oracle-database/19/sqlrf/Introduction-to-Oracle-SQL.html), [PostgreSQL](https://www.postgresql.org/), [MongoDB](https://www.mongodb.com/docs/kafka-connector/current/sink-connector/fundamentals/change-data-capture/), or [Cassandra](https://cassandra.apache.org/_/index.html). CDC works by continuously monitoring the database for any changes made to the data. Multiple [types of change data capture](https://www.fivetran.com/blog/change-data-capture-what-it-is-and-how-to-use-it#:~:text=Log%2Dbased%20change%20data%20capture) patterns can be used for data processing from a database. These include *log-based CDC, trigger-based CDC, CDC based on timestamps, and difference-based CDC*.

When you make operations like `INSERT`, `UPDATE`, or `DELETE,` against a database, a change is detected, and CDC captures the change and records it in a transaction log. The captured changes are then transformed into a format that can be consumed in real time by downstream systems. The downstream systems can be a search index, cache, stream analytics applications, messaging queues, or data warehouse. 

![How Change Data Capture (CDC) works with Streaming Database (3).png](https://static.apiseven.com/uploads/2023/04/07/4rhtYk0M_How%20Change%20Data%20Capture%20%28CDC%29%20works%20with%20Streaming%20Database%20%283%29.png)

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, as its shown in the diagram below:

![How Change Data Capture (CDC) works with Streaming Database (2).png](https://static.apiseven.com/uploads/2023/04/07/ERCS4RFH_How%20Change%20Data%20Capture%20%28CDC%29%20works%20with%20Streaming%20Database%20%282%29.png)

## Change Data Capture use cases

There are several use cases for streaming CDC where it shines. As we understood, CDC can detect changes to the database table as soon as any of the following operations `INSERT`, `UPDATE`, or `DELETE` is made. It opens doors for us to **process real-time analytics** on data.

CDC is commonly used for **data synchronization** between different data sources. For example, it can be used to **synchronize on-premises data to the cloud** or if an organization has multiple databases that need to be kept in sync, CDC can **continuously capture and propagate changes** between them. It can be used for **database replication** across different environments, such as from production to a staging environment. This ensures that the data in the staging environment is always up-to-date and consistent with the production environment.

Nowadays, many organizations are **adopting event-driven architectures** by implementing small microservices or a Function as Service (FaaS) that responds to specific changes in the system. In such cases, streaming CDC can help microservices communicate with each other by **[enabling real-time data sharing](https://debezium.io/blog/2019/02/19/reliable-microservices-data-exchange-with-the-outbox-pattern/)**.

## Change Data Capture with Streaming Database

A *streaming database* is a type of database that is designed to handle continuous data streams in real-time and makes it possible to query this data. You can read more about [how a Streaming database differs from a Traditional database](https://www.iambobur.com/post/how-streaming-database-differs-from-a-traditional-database) and [how to choose the right streaming database](https://www.iambobur.com/post/how-to-choose-the-right-streaming-database) in my other blog posts. CDC is particularly useful when working with streaming databases, you can **ingest CDC data from directly databases** (See an example in the next section) without setting up additional services like [Kafka](https://kafka.apache.org/). 

If you’re already using [Debezium](https://debezium.io/) to extract CDC logs into Kafka, you can just set up RisingWave to consume changes from that Kafka topic. In this case, Kafka acts like a hub of CDC data, and beside RisingWave, other downstream systems like search index or data warehouses can consume changes as well.

Also, the streaming database enables downstream systems to **stay in sync with the source database** and **have access to the latest changes** made to the data [using simple SQL queries](https://www.iambobur.com/post/query-real-time-data-in-kafka-using-sql). With the help of a streaming database, you can denormalize CDC data via a [materialized view](https://en.wikipedia.org/wiki/Materialized_view) where the results of a query (lookup data) are stored in the streaming database's local cache to improve performance. Also, you can query that data from [BI and data analytic platforms](https://www.risingwave.dev/docs/current/rw-integration-summary/#bi-and-data-analytic-platforms).

Streaming databases can extract data in real-time from different sources at the same time, and merge and [sink data to MySQL using a JDBC connector](https://www.risingwave.dev/docs/current/sink-to-mysql-with-jdbc/).

## Ingest CDC data from MySQL using the RisingWave connector

Suppose you are working for an e-commerce company that receives a large volume of orders every day. To ensure the timely fulfillment of these orders, you need to process them in real time as they come in.

To do this, you decide to use a streaming database such as [RisingWave](https://www.risingwave-labs.com/) that can capture changes to the order data stored in MySQL DB using its native [MySQL CDC connector](https://www.risingwave.dev/docs/current/ingest-from-mysql-cdc/). This way, you can read the order data from the database as soon as it's available, and process it in real-time to ensure fast and efficient order fulfillment.

With [RisingWave’s MySQL CDC connector](https://www.risingwave.dev/docs/current/ingest-from-mysql-cdc/), RisingWave can connect to MySQL directly to obtain data from its binary log (`binlog`) without other stream processing services and manage real-time a materialized view on top of CDC data.

This demo leverages the following [GitHub repository](https://github.com/risingwavelabs/risingwave/tree/main/integration_tests/mysql-cdc) where we assume that all necessary things are set up using [Docker compose](https://github.com/risingwavelabs/risingwave/blob/main/integration_tests/delivery/docker-compose.yml). It will start multiple containers including MySQL with pre-populated inputs. You can see [here](https://www.risingwave.dev/docs/current/risingwave-docker-compose/) what are the additional components in this Docker configuration. 

### Before You Begin

- Ensure you have Docker and Docker Compose installed in your environment.
- Clone the [GitHub repository](https://github.com/risingwavelabs/risingwave/tree/main/integration_tests/mysql-cdc), and run containers with `docker compose up` .
- Before using the native MySQL CDC connector in RisingWave, you need to complete several configurations on MySQL. See the [Set up MySQL section](https://www.risingwave.dev/docs/current/ingest-from-mysql-cdc/#set-up-mysql) for more details.
- We are going to use the [PostgreSQL interactive terminal](https://www.risingwave.dev/docs/current/install-psql-without-postgresql/) `psql` for running queries and retrieving results from RisingWave. Open your CLI, and run the following command to use the PostgreSQL developer tool: `psql -h localhost -p 4566 -d dev -U root`

### Create a Source/Table in RisingWave

For RisingWave to ingest CDC data from MySQL, you must create a table (`CREATE TABLE`) with primary keys and connector settings. 

```sql
create table orders (
    order_id int,
    order_date bigint,
    customer_name varchar,
    price decimal,
    product_id int,
    order_status smallint,
    PRIMARY KEY (order_id)
) with (
    connector = 'mysql-cdc',
    hostname = 'mysql',
    port = '3306',
    username = 'root',
    password = '123456',
    database.name = 'mydb',
    table.name = 'orders',
    server.id = '1'
);
```

### Create a materialized view

Let’s assume that we want to calculate **the total number of products ordered** up to now. Now you can think “_We can achieve the same even without using a CDC and streaming database_”. Well, you can do so if the Order dataset is static, it doesn’t change frequently and it is in small size. But in reality, there will be many orders, the data changes so frequently, and real-time access to this data is required. 

CDC detects newly entered orders quickly and **will update target systems with the latest data** in near real-time. Then the streaming database takes these changes made to the source table and stores them in a locally cached table called `orders` respectively. This way CDC data will be persisted in RisingWave for further analysis in real-time with the materialized views. Any materialized view defined on top of this source will be incrementally updated and product count will be calculated behind the scene as new change arrives from CDC.

Now you have the idea of using the CDC, streaming database, and materialized view to deal with fast-changing and rapidly growing orders data. 

We can use the following SQL query to create a new materialized view for the product count:

```sql
CREATE MATERIALIZED VIEW product_count AS
SELECT
    product_id,
    COUNT(*) as product_count
FROM
    orders
GROUP BY
    product_id;
```

### Run a query on the CDC materialized view

Now we can run a simple query on the materialized view we created in the previous step to see actual the number of products ordered by limiting the result to 10 rows:

```sql
SELECT
    *
FROM
    product_count;
```

Then you can see the output retrieved from Risingwave materialized view in its database:

```bash
product_id | product_count
------------+---------------
          1 |             1
          2 |             2
```

It does not end, now if you try to insert any new row to MySQL `mydb.orders` table, it is immediately reflected on the `product_count` materialized view in RisingWave.

```sql
insert into
  orders
values
  (4, 1558430840000, 'John', 10.50, 1, 1);
```

New output on `product_count` materialized view in real-time:

```sql
product_id | product_count
------------+---------------
          1 |             2
          2 |             2
```

## Takeaways

- CDC is capable of detecting changes to the database table, creating change events and later these events will be consumed in real-time by downstream systems.
- The streaming database acts as a stateful stream processor that can materialize the CDC events stream into a table that represents the current state.
- Compared to other relational databases, the streaming database uses a special streaming framework, instead of using a query engine to compute point-in-time results.
- A materialized view in a streaming database can be used to easily obtain, large amounts and rapidly changing data that are difficult to query directly.

### Related resources

- [how to choose the right streaming database](https://www.iambobur.com/post/how-to-choose-the-right-streaming-database)
- [How Streaming database differs from a Traditional database?](https://www.iambobur.com/post/how-streaming-database-differs-from-a-traditional-database)
- [Query Real-Time Data in Kafka Using SQL](https://www.iambobur.com/post/query-real-time-data-in-kafka-using-sql)

### Recommended content

- [Is RisingWave the Right Streaming Database?](https://www.risingwave-labs.com/blog/is-risingwave-the-right-streaming-database/#durability)
- [Rethinking Stream Processing and Streaming Databases](https://www.risingwave-labs.com/blog/Rethinking_stream_processing_and_streaming_databases/)

### Community

🙋 [Join the Risingwave Community](https://join.slack.com/t/risingwave-community/shared_invite/zt-120rft0mr-d8uGk3d~NZiZAQWPnElOfw)

### About the author

Visit my personal blog: [www.iambobur.com](https://www.iambobur.com/)