Skip to main content

Ingest data from MySQL CDC

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

  1. Create a MySQL user with the following query.
CREATE USER 'user'@'%' IDENTIFIED BY 'password';
  1. Grant the appropriate privileges to the user.
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user'@'%';
  1. Finalize the privileges.
FLUSH PRIVILEGES;

Enable the binlog

The binlog must be enabled for MySQL replication. The binary logs record transaction updates for replication tools to propagate changes.

  1. Check if the log-bin is already on.
SHOW VARIABLES LIKE 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | OFF |
+---------------+-------+
  1. 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.
server-id         = 223344
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 10
  1. Confirm your changes by checking the log-bin again.
SHOW VARIABLES LIKE 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | ON |
+---------------+-------+

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 table. Note that a primary key is required.

CREATE TABLE [ IF NOT EXISTS ] table_name (
column_name data_type PRIMARY KEY , ...
PRIMARY KEY ( column_name, ... )
)
WITH (
connector='mysql-cdc',
connector_parameter='value', ...
)
[ FORMAT DEBEZIUM ENCODE JSON ];

Syntax for creating a CDC source.

CREATE SOURCE [ IF NOT EXISTS ] source_name WITH (
connector='mysql-cdc',
<field>=<value>, ...
);

Connector parameters

All the fields listed below are required. Note that the value of these parameters should be enclosed in single quotation marks.

FieldNotes
hostnameHostname of the database.
portPort number of the database.
usernameUsername of the database.
passwordPassword of the database.
database.nameName of the database. Note that RisingWave cannot read data from a built-in MySQL database, such as mysql, sys, etc.
table.nameName of the table that you want to ingest data from.
server.idRequired 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.
transactionalOptional. 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 details, see Transaction within a CDC table.
snapshotOptional. 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.

Debezium parameters

Debezium v2.4 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.

CREATE SOURCE mysql_mydb WITH (
connector = 'mysql-cdc',
hostname = '127.0.0.1',
port = '8306',
username = 'root',
password = '123456',
database.name = 'mydb',
server.id = 5888,
debezium.schema.history.internal.skip.unparseable.ddl = 'true'
);

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.

Examples

Create a single CDC table

The following example creates a table in RisingWave that reads CDC data from the orders table in MySQL. When connecting to a specific table in MySQL, use the CREATE TABLE command.

CREATE TABLE orders (
order_id int,
order_date bigint,
customer_name string,
price decimal,
product_id int,
order_status smallint,
PRIMARY KEY (order_id)
) WITH (
connector = 'mysql-cdc',
hostname = '127.0.0.1',
port = '3306',
username = 'root',
password = '123456',
database.name = 'mydb',
table.name = 'orders',
server.id = '5454'
);

Create multiple CDC tables with the same source

RisingWave supports creating a single MySQL source that allows you to read CDC data from multiple tables located in the same database.

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.

CREATE SOURCE mysql_mydb WITH (
connector = 'mysql-cdc',
hostname = '127.0.0.1',
port = '8306',
username = 'root',
password = '123456',
database.name = 'mydb',
server.id = 5888
);

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.

CREATE TABLE t1_rw (
v1 int,
v2 int,
PRIMARY KEY(v1)
) FROM mysql_mydb TABLE 'mydb.t1';

You can create another CDC table in RisingWave that ingests data from table t3 in the same database mydb.

CREATE TABLE t3_rw (
v1 INTEGER,
v2 timestamptz,
PRIMARY KEY (v1)
) FROM mysql_mydb TABLE 'mydb.t3';

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.

SELECT * FROM __internal_orders_rw_4002_streamcdcscan_5002;

-[ RECORD 1 ]-----+---------------------------------------------------------------
split_id | 5001
o_orderkey | 4024320
backfill_finished | f
row_count | 1006080
cdc_offset | {"MySql": {"filename": "binlog.000005", "position": 60946679}}

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 typeRisingWave type
BOOLEAN, BOOLBOOLEAN
BIT(1)BOOLEAN*
BIT(>1)No support
TINYINTSMALLINT
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
TINYBLOBBYTEA
TINYTEXTCHARACTER VARYING
BLOBBYTEA
TEXTCHARACTER VARYING
MEDIUMBLOBBYTEA
MEDIUMTEXTCHARACTER VARYING
LONGBLOBBYTEA
LONGTEXTBYTEA or CHARACTER VARYING
JSONJSONB
ENUMCHARACTER VARYING*
SETNo support
YEAR[(2|4)]INTEGER
TIMESTAMP[(M)]TIMESTAMPTZ
DATEDATE
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 typeRisingWave typeMySQL rangeRisingWave range
TIMETIME-838:59:59.000000 to 838:59:59.00000000:00:00 to 23:59:59
DATEDATE1000-01-01 to 9999-12-310001-01-01 to 9999-12-31
DATETIMETIMESTAMP1000-01-01 00:00:00.000000 to 9999-12-31 23:59:59.499991973-03-03 09:46:40 to 5138-11-16 09:46:40
TIMESTAMPTIMESTAMPTZ1970-01-01 00:00:01.000000 to 2038-01-19 03:14:07.4999990001-01-01 00:00:00 to 9999-12-31 23:59:59

Help us make this doc better!