> ## Documentation Index > Fetch the complete documentation index at: https://docs.risingwave.com/llms.txt > Use this file to discover all available pages before exploring further. # MySQL CDC > Replicate data from MySQL to RisingWave in real time using native CDC. Supports shared sources for multi-table ingestion and AWS RDS. No Kafka or Debezium required. Connect RisingWave to a MySQL database to ingest data changes in real time using the native MySQL CDC source connector. Need help generating SQL? Use [Claude Code](https://claude.ai/claude-code) or [Cursor](https://cursor.com) with the [RisingWave MCP server](https://github.com/risingwavelabs/risingwave-mcp) to generate and run SQL interactively. The native MySQL CDC connector supports MySQL versions 5.7, 8.0, and 8.4, as well as compatible databases such as MariaDB and TiDB. ## Prerequisites Before using the native MySQL CDC connector in RisingWave, you need to configure your MySQL database properly. * [Self-hosted MySQL](/ingestion/sources/mysql/self-hosted) * [AWS RDS MySQL or Aurora](/ingestion/sources/mysql/aws-rds) ## Connect to MySQL To ingest CDC data from MySQL, you first create a shared source using the `CREATE SOURCE` statement. This source establishes the connection to the MySQL database. Then, for each upstream table you want to ingest, you define a corresponding table in RisingWave using the `CREATE TABLE FROM SOURCE` statement. ### Create a shared source Use the `CREATE SOURCE` statement to create a shared source. ```sql theme={null} CREATE SOURCE [ IF NOT EXISTS ] WITH ( connector='mysql-cdc', =, ... ); ``` ### Create a table from the shared source Next, create a table from the shared source to ingest data from a specific upstream table. In the `CREATE TABLE` statement, you must define a primary key that matches the upstream table's primary key and specify the upstream table's name. ```sql theme={null} CREATE TABLE [ IF NOT EXISTS ] ( PRIMARY KEY , ... PRIMARY KEY ( , ... ) ) [ INCLUDE timestamp AS ] WITH ( snapshot='true' ) FROM TABLE ; ``` ## Parameters Use the these parameters when creating a shared source for MySQL CDC. | Parameter | Description | | :------------------- | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `connector` | Must be set to `mysql-cdc`. | | `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. | | `auto.schema.change` | **Optional**. Specify whether you want to enable replicating MySQL table schema change. Set `auto.schema.change = 'true'` to enable it. | | `ssl.mode` | **Optional**. The `ssl.mode` parameter determines the level of SSL/TLS encryption for secure communication with MySQL. Accepted values are `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. | Use these parameters when creating a table from a shared source. | Parameter | Description | | :------------------------------ | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 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. | | backfill.parallelism | **Optional**. Controls the parallelism of CDC table backfill. When set to a value greater than 0, enables parallelized CDC backfill. The default value is `0` (disabled). | | backfill\_num\_rows\_per\_split | **Optional**. Specifies number of rows per split for parallelized CDC backfill. Only effective when `backfill.parallelism > 0`. The default value is `100000`. | | backfill\_as\_even\_splits | **Optional**. Whether to create even splits for parallelized CDC backfill. Only effective when `backfill.parallelism > 0`. The default value is `true`. | For large tables, you can significantly speed up the initial data load by enabling parallelized backfill. Configure this feature using the `backfill.parallelism`, `backfill_num_rows_per_split`, and `backfill_as_even_splits` parameters. ```sql theme={null} CREATE TABLE large_table ( id integer primary key, data varchar ) WITH ( backfill.parallelism = '4', backfill.num_rows_per_split = '50000', backfill.as_even_splits = 'true' ) FROM mysql_db TABLE 'public.large_table'; ``` ## Debezium parameters [Debezium v2.6 connector configuration properties](https://debezium.io/documentation/reference/2.6/connectors/mysql.html#mysql-advanced-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`. ```sql theme={null} 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](https://debezium.io) 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. | | 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`, `table_name`) to provide contextual information about where the data resides within the MySQL database. ```sql theme={null} CREATE TABLE person ( id int, name varchar, email_address varchar, credit_card varchar, city varchar, PRIMARY KEY (id) ) INCLUDE TIMESTAMP AS commit_ts INCLUDE DATABASE_NAME as database_name INCLUDE TABLE_NAME as table_name FROM mysql_source TABLE 'public.person'; ``` ## Examples Connect to the upstream database by creating a CDC source using the [CREATE SOURCE](/sql/commands/sql-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. ```sql theme={null} 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. ```sql theme={null} CREATE TABLE t1_rw ( v1 int, v2 int, PRIMARY KEY(v1) ) FROM mysql_mydb TABLE 'mydb.t1'; ``` You can also create another CDC table in RisingWave that ingests data from table `t3` in the same database `mydb`. ```sql theme={null} 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](/sql/commands/sql-show-internal-tables) command and query from it. For instance, the following SQL query shows the progress of a CDC table named `orders_rw`. ```sql theme={null} 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](/sql/data-types/overview). 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) | BYTEA | | TINYINT | SMALLINT | | SMALLINT\[(M)] | SMALLINT | | MEDIUMINT\[(M)] | INTEGER | | INT, INTEGER\[(M)] | INTEGER | | BIGINT\[(M)] | BIGINT | | REAL\[(M,D)] | REAL | | FLOAT\[(P)] | REAL | | 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 | Not supported | 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](/integrations/other/dbt#define-dbt-models). First, we create a `source` model `mysql_mydb.sql`. ```sql theme={null} {{ config(materialized='source') }} CREATE SOURCE {{ this }} WITH ( connector = 'mysql-cdc', hostname = '127.0.0.1', port = '8306', username = 'root', password = '123456', database.name = 'mydb', server.id = 5888 ); ``` And then we create a `table_with_connector` model `t1_rw.sql`. ``` {{ config(materialized='table_with_connector') }} CREATE TABLE {{ this }} ( v1 int, v2 int, PRIMARY KEY(v1) ) FROM {{ ref('mysql_mydb') }} TABLE 'mydb.t1'; ``` ## Automatically map upstream table schema 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: ```sql theme={null} CREATE TABLE supplier (*) FROM mysql_source TABLE 'public.supplier'; ``` And this it the output of `DESCRIBE supplier;` ``` Name | Type | Is Hidden | Description -------------------+-------------------+-----------+------------- s_suppkey | bigint | false | s_name | character varying | false | s_address | character varying | false | s_nationkey | bigint | false | s_phone | character varying | false | s_acctbal | numeric | false | s_comment | character varying | false | primary key | s_suppkey | | distribution key | s_suppkey | | table description | supplier | | (10 rows) ``` ## Auto schema change **PREMIUM FEATURE** This is a premium feature. For a comprehensive overview of all premium features and their usage, please see [RisingWave premium features](/get-started/premium-features). 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: ```SQL theme={null} CREATE SOURCE mysql_source WITH ( connector = 'mysql-cdc', hostname = 'localhost', port = '3306', username = 'root', password = 'your_password', database.name = 'mytest', server.id = '5701', auto.schema.change = 'true' ); ``` Create a RisingWave table from the MySQL source: ```SQL theme={null} CREATE TABLE rw_customers ( id BIGINT, modified TIMESTAMP, custinfo JSONB, PRIMARY KEY (id) ) FROM mysql_source TABLE 'mytest.customers'; ``` Add columns to the MySQL table and observe the changes in RisingWave: ```SQL theme={null} -- In MySQL: USE mytest; ALTER TABLE customers ADD COLUMN v1 VARCHAR(255); ALTER TABLE customers ADD COLUMN v2 DOUBLE(5,2); -- In RisingWave (after a brief pause): DESCRIBE rw_customers; ``` And this it the output of `DESCRIBE rw_customers;` ``` Name | Type | Is Hidden | Description ----------------------------+--------------------------------+-----------+------------- id | bigint | false | NULL modified | timestamp without time zone | false | NULL custinfo | jsonb | false | NULL v1 | character varying | false | NULL v2 | double precision | false | NULL primary key id | NULL | NULL | NULL distribution key id | NULL | NULL | NULL table description rw_customers | NULL | NULL | NULL ``` ## Expression as a column RisingWave allows users to define expressions as table columns. For example, in the SQL statement below, `next_id` is not a column from the source MySQL table. Instead, it is a generated column that RisingWave computes dynamically while ingesting data. The value of `next_id` for each row is always equal to `id + 1`: ```sql theme={null} CREATE TABLE person ( id integer PRIMARY KEY, name varchar, next_id int AS id + 1, PRIMARY KEY (id) ) FROM mysql_mydb TABLE 'mydb.person'; ``` Currently, generated columns must appear at the end of the schema definition. If a column from the upstream source appears after a generated column, RisingWave will return an error. For example, the following statement will fail because `name`, an upstream column, is placed after the generated column `next_id`: ```sql theme={null} CREATE TABLE person ( id integer PRIMARY KEY, next_id int AS id + 1, name varchar, PRIMARY KEY (id) ) FROM mysql_mydb TABLE 'mydb.person'; ``` To avoid errors, ensure that all generated columns are positioned at the end of the schema definition. ## 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: 1. Create a table to backfill historical data: ``` CREATE TABLE t3 (id INTEGER, v1 TIMESTAMP WITH TIME ZONE, PRIMARY KEY(id)) FROM mysql_source TABLE 'mydb.t3'; ``` 2. List the internal tables to find the relevant backfill executor state: ``` SHOW INTERNAL TABLES; ``` Output: ``` Name --------------------------------- __internal_t3_3_streamcdcscan_4 __internal_mysql_source_1_source_2 (2 rows) ``` 3. Check the internal state of the backfill executor: ``` SELECT * FROM __internal_t3_3_streamcdcscan_4; ``` Output: ``` split_id | id | backfill_finished | row_count | cdc_offset ----------+----+-------------------+-----------+-------------------------------------------------- 3 | 5 | t | 4 | {"MySql": {"filename": "binlog.000067", "position": 39101}} (1 row) ``` ### 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): ```SQL theme={null} SHOW MASTER STATUS; ``` Then compare the above offset with source offset stored in the state table to determine the CDC progress.