Ingest CDC data from MySQL
Change Data Capture (CDC) refers to the process of identifying and capturing data changes in a database, and then delivering the changes to a downstream service in real time.
RisingWave supports ingesting row-level data (INSERT
, UPDATE
, and DELETE
operations) from the changes of a MySQL database. The supported MySQL versions are 5.7 and 8.0.x.
You can ingest CDC data from MySQL in two ways:
- Using the native MySQL CDC connector in RisingWave With this connector, RisingWave can connect to MySQL databases directly to obtain data from the binlog without starting additional services.
- Using a CDC tool and a message broker You can use a CDC tool and then use the Kafka, Pulsar, or Kinesis connector to send the CDC data to RisingWave.
This topic describes how to ingest MySQL CDC data into RisingWave using the native MySQL CDC connector. Using an external CDC tool and a message broker is introduced in Create source via event streaming systems.
Set up MySQL
Before using the native MySQL CDC connector in RisingWave, you need to complete several configurations on MySQL.
To use the MySQL CDC features, we need to create a MySQL user account with appropriate privileges on all databases for which RisingWave will read from.
Create a user and grant privileges
- Create a MySQL user with the following query.
- Grant the appropriate privileges to the user.
- Finalize the privileges.
Enable the binlog
The binlog must be enabled for MySQL replication. The binary logs record transaction updates for replication tools to propagate changes.
- Check if the
log-bin
is already on.
- If it is
OFF
, configure your MySQL server configuration file, my.cnf, with the following properties described below. Restart your MySQL server to let the configurations take effect.
- Confirm your changes by checking the
log-bin
again.
See Setting up MySQL for more details.
Notes about running RisingWave from binaries
If you are running RisingWave locally from binaries and intend to use the native CDC source connectors or the JDBC sink connector, make sure that you have JDK 11 or later versions installed in your environment.
Create a table using the native CDC connector in RisingWave
To ensure all data changes are captured, you must create a table and specify primary keys. See the CREATE TABLE command for more details.
Syntax
Syntax for creating a CDC source.
Syntax for creating a CDC table. Note that a primary key is required and must be consistent with the upstream table.
Connector parameters
All the fields listed below are required. Note that the value of these parameters should be enclosed in single quotation marks.
Field | Notes |
---|---|
hostname | Hostname of the database. |
port | Port number of the database. |
username | Username of the database. |
password | Password of the database. |
database.name | Name of the database. Note that RisingWave cannot read data from a built-in MySQL database, such as mysql, sys, etc. |
table.name | Name of the table that you want to ingest data from. |
server.id | Required if creating a shared source. A numeric ID of the database client. It must be unique across all database processes that are running in the MySQL cluster. If not specified, RisingWave will generate a random ID. |
ssl.mode | Optional. The ssl.mode parameter determines the level of SSL/TLS encryption for secure communication with MySQL. It accepts three values: disabled, preferred, and required. The default value is disabled. When set to required, it enforces TLS for establishing a connection. |
transactional | Optional. Specify whether you want to enable transactions for the CDC table that you are about to create. By default, the value is ‘true’ for shared sources, and ‘false’ otherwise. This feature is also supported for shared CDC sources for multi-table transactions. For performance considerations, transactions involving changes to more than 4096 rows cannot be guaranteed. |
The following fields are used when creating a CDC table.
Field | Notes |
---|---|
snapshot | Optional. If false, CDC backfill will be disabled and only upstream events that have occurred after the creation of the table will be consumed. This option can only be applied for tables created from a shared source. |
snapshot.interval | Optional. Specifies the barrier interval for buffering upstream events. The default value is 1. |
snapshot.batch_size | Optional. Specifies the batch size of a snapshot read query from the upstream table. The default value is 1000. |
Regarding the INCLUDE timestamp AS column_name
clause, it allows you to ingest the upstream commit timestamp. For historical data, the commit timestamp will be set to 1970-01-01 00:00:00+00:00
. Here is an example:
You can see the INCLUDE clause for more details.
Debezium parameters
Debezium v2.6 connector configuration properties can also be specified under the WITH
clause when creating a table or shared source. Add the prefix debezium.
to the connector property you want to include.
For instance, to skip unknown DDL statements, specify the schema.history.internal.skip.unparseable.ddl
parameter as debezium.schema.history.internal.skip.unparseable.ddl
.
Data format
Data is in Debezium JSON format. Debezium is a log-based CDC tool that can capture row changes from various database management systems such as PostgreSQL, MySQL, and SQL Server and generate events with consistent structures in real time. The MySQL CDC connector in RisingWave supports JSON as the serialization format for Debezium data. The data format does not need to be specified when creating a table with mysql-cdc
as the source.
Metadata options
Below are the metadata columns available for MySQL CDC.
Field | Notes |
---|---|
database_name | Name of the database. |
schema_name | Name of the schema. |
table_name | Name of the table. |
For instance, the person table below contains columns for typical personal information. It also includes metadata fields (database_name
, schema_name
, table_name
) to provide contextual information about where the data resides within the MySQL database.
Examples
Connect to the upstream database by creating a CDC source using the CREATE SOURCE command and MySQL CDC parameters. The data format is fixed as FORMAT PLAIN ENCODE JSON
so it does not need to be specified.
With the source created, you can create multiple CDC tables that ingest data from different tables in the upstream database without needing to specify the database connection parameters again.
For instance, the following CDC table in RisingWave ingests data from table t1
in the database mydb
. When specifying the MySQL table name in the FROM
clause after the keyword TABLE
, the database name must also be specified.
You can also create another CDC table in RisingWave that ingests data from table t3
in the same database mydb
.
To check the progress of backfilling historical data, find the corresponding internal table using the SHOW INTERNAL TABLES command and query from it. For instance, the following SQL query shows the progress of a CDC table named orders_rw
.
Data type mapping
The following table shows the corresponding data type in RisingWave that should be specified when creating a source. For details on native RisingWave data types, see Overview of data types.
RisingWave data types marked with an asterisk indicate that while there is no corresponding RisingWave data type, the ingested data can still be consumed as the listed type.
MySQL type | RisingWave type |
---|---|
BOOLEAN, BOOL | BOOLEAN |
BIT(1) | BOOLEAN* |
BIT(>1) | No support |
TINYINT | SMALLINT |
SMALLINT[(M)] | SMALLINT |
MEDIUMINT[(M)] | INTEGER |
INT, INTEGER[(M)] | INTEGER |
BIGINT[(M)] | BIGINT |
REAL[(M,D)] | REAL |
FLOAT[(P)] | REAL |
FLOAT(M,D) | DOUBLE PRECISION |
DOUBLE[(M,D)] | DOUBLE PRECISION |
CHAR[(M)] | CHARACTER VARYING |
VARCHAR[(M)] | CHARACTER VARYING |
BINARY[(M)] | BYTEA |
VARBINARY[(M)] | BYTEA |
TINYBLOB | BYTEA |
TINYTEXT | CHARACTER VARYING |
BLOB | BYTEA |
TEXT | CHARACTER VARYING |
MEDIUMBLOB | BYTEA |
MEDIUMTEXT | CHARACTER VARYING |
LONGBLOB | BYTEA |
LONGTEXT | BYTEA or CHARACTER VARYING |
JSON | JSONB |
ENUM | CHARACTER VARYING* |
SET | No support |
YEAR[(2|4)] | INTEGER |
TIMESTAMP[(M)] | TIMESTAMPTZ |
DATE | DATE |
TIME[(M)] | TIME |
DATETIME[(fsp)] Optional fractional seconds precision (fsp: 0-6). If omitted, the default precision is 0. | TIMESTAMP |
NUMERIC[(M[,D])] | NUMERIC |
DECIMAL[(M[,D])] | NUMERIC |
GEOMETRY, LINESTRING, POLYGON, MULTIPOINT, MULTILINESTRING, MULTIPOLYGON, GEOMETRYCOLLECTION | STRUCT |
Please be aware that the range of specific values varies among MySQL types and RisingWave types. Refer to the table below for detailed information.
MySQL type | RisingWave type | MySQL range | RisingWave range |
---|---|---|---|
TIME | TIME | -838:59:59.000000 to 838:59:59.000000 | 00:00:00 to 23:59:59 |
DATE | DATE | 1000-01-01 to 9999-12-31 | 0001-01-01 to 9999-12-31 |
DATETIME | TIMESTAMP | 1000-01-01 00:00:00.000000 to 9999-12-31 23:59:59.49999 | 1973-03-03 09:46:40 to 5138-11-16 09:46:40 |
TIMESTAMP | TIMESTAMPTZ | 1970-01-01 00:00:01.000000 to 2038-01-19 03:14:07.499999 | 0001-01-01 00:00:00 to 9999-12-31 23:59:59 |
Use dbt to ingest data from MySQL CDC
Here is an example of how to use dbt to ingest data from MySQL CDC. In this dbt example, source
and table_with_connector
models will be used. For more details about these two models, please refer to Use dbt for data transformations.
First, we create a source
model mysql_mydb.sql
.
And then we create a table_with_connector
model t1_rw.sql
.
Automatically map upstream table schema
PREMIUM EDITION FEATURE
This feature is only available in the premium edition of RisingWave. The premium edition offers additional advanced features and capabilities beyond the free and community editions. If you have any questions about upgrading to the premium edition, please contact our sales team at sales@risingwave-labs.com.
PUBLIC PREVIEW
This feature is in the public preview stage, meaning it’s nearing the final product but is not yet fully stable. If you encounter any issues or have feedback, please contact us through our Slack channel. Your input is valuable in helping us improve the feature. For more information, see our Public preview feature list.
RisingWave supports automatically mapping the upstream table schema when creating a CDC table from a MySQL CDC source. Instead of defining columns individually, you can use *
when creating a table to ingest all columns from the source table. Note that *
cannot be used if other columns are specified in the table creation process.
Below is an example to create a table that ingests all columns from the upstream table from the MySQL database:
And this it the output of DESCRIBE supplier;
Automatically change schema
PREMIUM EDITION FEATURE
This feature is only available in the premium edition of RisingWave. The premium edition offers additional advanced features and capabilities beyond the free and community editions. If you have any questions about upgrading to the premium edition, please contact our sales team at sales@risingwave-labs.com.
PUBLIC PREVIEW
This feature is in the public preview stage, meaning it’s nearing the final product but is not yet fully stable. If you encounter any issues or have feedback, please contact us through our Slack channel. Your input is valuable in helping us improve the feature. For more information, see our Public preview feature list.
RisingWave supports auto schema changes in MySQL CDC. It ensures that your RisingWave pipeline stays synchronized with any schema changes in the source database, reducing the need for manual updates and preventing inconsistencies.
Currently, RisingWave supports the ALTER TABLE
command with the following operations, and we plan to add support for additional DDL operations in future releases.
ADD COLUMN [DEFAULT expr]
: Allows you to add a new column to an existing table. Only constant value expressions are supported for the default value.DROP COLUMN
: Allows you to remove an existing column from a table.
To enable this feature, set auto.schema.change = 'true'
in your MySQL CDC source configuration:
Create a RisingWave table from the MySQL source:
Add columns to the MySQL table and observe the changes in RisingWave:
And this it the output of DESCRIBE rw_customers;
Monitor the progress of direct CDC
To observe the progress of direct CDC for MySQL, use the following methods:
For historical data
Historical data needs to be backfilled into the table. You can check the internal state of the backfill executor as follows:
- Create a table to backfill historical data:
- List the internal tables to find the relevant backfill executor state:
Output:
- Check the internal state of the backfill executor:
Output:
For real-time data
RisingWave stores source offset in the internal state table of source executor. You can check the current consumed offset by checking this table and comparing it with the upstream database’s log offset.
To get the current binlog offset, run this SQL query on upstream MySQL (earlier than 8.4.0):
Then compare the above offset with source offset stored in the state table to determine the CDC progress.
Was this page helpful?