--- title: "使用 Flink CDC 从 MySQL 数据库同步数据到 OceanBase 数据库 - OceanBase 数据库 V4.6.0 | OceanBase 文档中心" description: 使用 Flink CDC 从 MySQL 数据库同步数据到 OceanBase 数据库 Flink CDC (CDC Connectors for Apache Flink) 是 Apache Flink 的一组 Source 连接器，它支持从大多数据库中实时地读取存量历史数据和增量变更数据。Flink CDC 能够… --- 切换语言 - 中文站 - 简体中文 - International - English - 日本站 - 日本語 文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 4.6.0 # 使用 Flink CDC 从 MySQL 数据库同步数据到 OceanBase 数据库 更新时间：2026-05-20 14:02:53 [编辑](https://github.com/oceanbase/oceanbase-doc/edit/V4.6.0/zh-CN/500.data-migration/200.migrate-data-from-mysql-database-to-oceanbase-database/600.use-flink-cdc-to-migrate-data-from-mysql-database-to-oceanbase-database.md) Flink CDC (CDC Connectors for Apache Flink) 是 Apache Flink 的一组 Source 连接器，它支持从大多数据库中实时地读取存量历史数据和增量变更数据。Flink CDC 能够将数据库的全量和增量数据同步到消息队列和数据仓库中。Flink CDC 也可以用于实时数据集成，您可以使用它将数据库数据实时导入数据湖或者数据仓库。同时，Flink CDC 还支持数据加工，您可以通过它的 SQL Client 对数据库数据做实时关联、打宽、聚合，并将结果写入到各种存储中。CDC （Change Data Capture，即变更数据捕获）能够帮助您监测并捕获数据库的变动。CDC 提供的数据可以做很多事情，比如：做历史库、做近实时缓存、提供给消息队列（MQ），用户消费 MQ 做分析和审计等。 以下将介绍使用 Flink CDC 从 MySQL 数据库同步数据到 OceanBase 数据库。 ## Flink CDC 环境准备 下载 Flink 和所需要的依赖包： 1. 通过 [下载地址](https://downloads.apache.org/flink/) 下载 Flink。本文档使用的是 Flink 1.15.3，并将其解压至目录 `/FLINK_HOME/flink-1.15.3`。 2. 下载下面列出的依赖包，并将它们放到目录 `/FLINK_HOME/flink-1.15.3/lib/` 下。 - [flink-sql-connector-mysql-cdc-2.1.1.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.1.1/flink-sql-connector-mysql-cdc-2.1.1.jar) - [flink-connector-jdbc-1.15.3.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc/1.15.3/flink-connector-jdbc-1.15.3.jar) - [mysql-connector-java-5.1.47.jar](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/5.1.47/mysql-connector-java-5.1.47.jar) ## 准备数据 ### 准备 MySQL 数据库数据 在 MySQL 数据库中准备测试数据，作为导入 OceanBase 数据库的源数据。 1. 进入 MySQL 数据库。 ```bash [xxx@xxx /...] $mysql -hxxx.xxx.xxx.xxx -P3306 -uroot -p****** MySQL [(none)]> ``` 2. 创建数据库 `test_mysql_to_ob`，表 `tbl1` 和 `tbl2`，并插入数据。 ```sql MySQL [(none)]> CREATE DATABASE test_mysql_to_ob; Query OK, 1 row affected MySQL [(none)]> USE test_mysql_to_ob; Database changed MySQL [test_mysql_to_ob]> CREATE TABLE tbl1(col1 INT PRIMARY KEY, col2 VARCHAR(20),col3 INT); Query OK, 0 rows affected MySQL [test_mysql_to_ob]> INSERT INTO tbl1 VALUES(1,'China',86),(2,'Taiwan',886),(3,'Hong Kong',852),(4,'Macao',853),(5,'North Korea',850); Query OK, 5 rows affected Records: 5 Duplicates: 0 Warnings: 0 MySQL [test_mysql_to_ob]> CREATE TABLE tbl2(col1 INT PRIMARY KEY,col2 VARCHAR(20)); Query OK, 0 rows affected MySQL [test_mysql_to_ob]> INSERT INTO tbl2 VALUES(86,'+86'),(886,'+886'),(852,'+852'),(853,'+853'),(850,'+850'); Query OK, 5 rows affected Records: 5 Duplicates: 0 Warnings: 0 ``` ### 准备 OceanBase 数据库数据 在 OceanBase 数据库中创建存放源数据的表。 1. 登录 OceanBase 数据库。 使用 `user001` 用户登录集群的 `mysql001` 租户。 ```bash [xxx@xxx /...] $obclient -h10.10.10.2 -P2881 -uuser001@mysql001 -p -A Enter password: Welcome to the OceanBase. Commands end with ; or \g. Your OceanBase connection id is 3221536981 Server version: OceanBase 4.0.0.0 (r100000302022111120-7cef93737c5cd03331b5f29130c6e80ac950d33b) (Built Nov 11 2022 20:38:33) Copyright (c) 2000, 2018, OceanBase and/or its affiliates. All rights reserved. Type 'help;' or '\h' for help. Type '\c' to clear the current input statement. obclient [(none)]> ``` 2. 创建数据库 `test_mysql_to_ob` 和表 `mysql_tbl1_and_tbl2`。 ```sql obclient [(none)]> CREATE DATABASE test_mysql_to_ob; Query OK, 1 row affected obclient [(none)]> USE test_mysql_to_ob; Database changed obclient [test_mysql_to_ob]> CREATE TABLE mysql_tbl1_and_tbl2(col1 INT PRIMARY KEY,col2 INT,col3 VARCHAR(20),col4 VARCHAR(20)); Query OK, 0 rows affected ``` ## 启动 Flink 集群和 Flink SQL CLI 1. 使用下面的命令跳转至 Flink 目录下。 ```bash [xxx@xxx /FLINK_HOME] #cd flink-1.15.3 ``` 2. 使用下面的命令启动 Flink 集群。 ```bash [xxx@xxx /FLINK_HOME/flink-1.15.3] #./bin/start-cluster.sh ``` 启动成功的话，可以在 `http://localhost:8081/` 访问到 Flink Web UI，如下所示： ![Flink_Web_UI](https://obbusiness-private.oss-cn-shanghai.aliyuncs.com/doc/img/observer-enterprise/V4.0.0/data-migration/Flink_CDC/1Flink_Web_UI.png) #### 说明 执行 `./bin/start-cluster.sh` 后，如果提示：`bash: ./bin/start-cluster.sh: Permission denied`。需要把 `flink-1.15.3` 目录下的所有 `-rw-rw-r--` 权限的文件的权限都设置为 `-rwxrwxrwx` 权限。 示例如下： ```bash [xxx@xxx /FLINK_HOME/flink-1.15.3] # chmod -R 777 /FLINK_HOME/flink-1.15.3/* ``` 3. 使用下面的命令启动 Flink SQL CLI。 ```bash [xxx@xxx /FLINK_HOME/flink-1.15.3] #./bin/sql-client.sh ``` 启动成功后，可以看到如下的页面： ![Flink_SQL_CLI](https://obbusiness-private.oss-cn-shanghai.aliyuncs.com/doc/img/observer-enterprise/V4.0.0/data-migration/Flink_CDC/2Flink_SQL_CLI.png) ### 设置 checkpoint 在 Flink SQL CLI 中开启 `checkpoint`，每隔 `3` 秒做一次 `checkpoint`。 ```Flink Flink SQL> SET execution.checkpointing.interval = 3s; [INFO] Session property has been set. ``` ### 创建 MySQL CDC 表 在 Flink SQL CLI 中创建 MySQL 数据库对应的表。 对于 MySQL 数据库中 `test_mysql_to_ob` 的表 `tbl1` 和 `tbl2` 使用 Flink SQL CLI 创建对应的表，用于同步这些底层数据库表的数据。 ```Flink Flink SQL> CREATE TABLE mysql_tbl1 ( col1 INT PRIMARY KEY, col2 VARCHAR(20), col3 INT) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'xxx.xxx.xxx.xxx', 'port' = '3306', 'username' = 'root', 'password' = '******', 'database-name' = 'test_mysql_to_ob', 'table-name' = 'tbl1'); [INFO] Execute statement succeed. Flink SQL> CREATE TABLE mysql_tbl2 (col1 INT PRIMARY KEY, col2 VARCHAR(20)) WITH ('connector' = 'mysql-cdc', 'hostname' = 'xxx.xxx.xxx.xxx', 'port' = '3306', 'username' = 'root', 'password' = '******', 'database-name' = 'test_mysql_to_ob', 'table-name' = 'tbl2'); [INFO] Execute statement succeed. ``` 有关 MySQL CDC Connector WITH 选项的详细信息，请参见 [Connector Options](https://ververica.github.io/flink-cdc-connectors/release-2.2/content/connectors/mysql-cdc.html#connector-options)。 ### 创建 OceanBase CDC 表 在 Flink SQL CLI 中创建 OceanBase 数据库对应的表。创建 `mysql_tbl1_and_tbl2` 表，用来将关联后的数据写入 OceanBase 数据库中。 ```Flink Flink SQL> CREATE TABLE mysql_tbl1_and_tbl2( col1 INT PRIMARY KEY, col2 INT,col3 VARCHAR(20), col4 VARCHAR(20)) WITH ('connector' = 'jdbc', 'url' = 'jdbc:mysql://10.10.10.2:2881/test_mysql_to_ob', 'username' = 'root@mysql001', 'password' = '******', 'table-name' = 'mysql_tbl1_and_tbl2'); [INFO] Execute statement succeed. ``` 有关 JDBC SQL Connector WITH 选项的详细信息，请参见 [Connector Options](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/jdbc/#connector-options)。 ### 在 Flink SQL CLI 中将数据写入 OceanBase 数据库中 使用 Flink SQL 将表 tbl1 与表 tbl2 关联，并将关联后的信息写入 OceanBase 数据库中。 ```Flink Flink SQL> INSERT INTO mysql_tbl1_and_tbl2 SELECT t1.col1,t1.col3,t1.col2,t2.col2 FROM mysql_tbl1 t1,mysql_tbl2 t2 WHERE t1.col3=t2.col1; [INFO] Submitting SQL update statement to the cluster... Loading class `com.mysql.jdbc.Driver'. This is deprecated. The new driver class is `com.mysql.cj.jdbc.Driver'. The driver is automatically registered via the SPI and manual loading of the driver class is generally unnecessary. [INFO] SQL update statement has been successfully submitted to the cluster: Job ID: c5ee92498addf813858e448ec25e85af ``` #### 说明 本文档测试示例使用的 MySQL 驱动（`com.mysql.jdbc.Driver`）是 MySQL Connector/J 5.1.47 版本。新版本 MySQL 驱动（`com.mysql.cj.jdbc.Driver`）请使用 MySQL Connector/J 8.x 版本。 ## 查看关联数据写入 OceanBase 数据库情况 登录 OceanBase 数据数，在 `test_mysql_to_ob` 库中查看表 `mysql_tbl1_and_tbl2` 的数据。 ```sql obclient [test_mysql_to_ob]> SELECT * FROM mysql_tbl1_and_tbl2; +------+------+-------------+------+ | col1 | col2 | col3 | col4 | +------+------+-------------+------+ | 1 | 86 | China | +86 | | 2 | 886 | Taiwan | +886 | | 3 | 852 | Hong Kong | +852 | | 4 | 853 | Macao | +853 | | 5 | 850 | North Korea | +850 | +------+------+-------------+------+ 5 rows in set ``` ## 查看数据更新情况 1. 在 MySQL 数据库的表 `tbl1` 和 `tbl2` 中插入分别插入一条数据。 ```sql MySQL [test_mysql_to_ob]> INSERT INTO tbl1 VALUES(6,'code',673); Query OK, 1 row affected MySQL [test_mysql_to_ob]> INSERT INTO tbl2 VALUES(673,'+673'); Query OK, 1 row affected ``` 2. 在 OceanBase 数据库中查看数据是否同步。 ```sql obclient [test_mysql_to_ob]> SELECT * FROM mysql_tbl1_and_tbl2; +------+------+-------------+------+ | col1 | col2 | col3 | col4 | +------+------+-------------+------+ | 1 | 86 | China | +86 | | 2 | 886 | Taiwan | +886 | | 3 | 852 | Hong Kong | +852 | | 4 | 853 | Macao | +853 | | 5 | 850 | North Korea | +850 | | 6 | 673 | code | +673 | +------+------+-------------+------+ 6 rows in set ``` 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线