使用 Flink CDC 从 OceanBase 数据库迁移数据到 MySQL 数据库

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 从 OceanBase 数据库迁移数据到 MySQL 数据库。

环境准备

配置 OceanBase Binlog 服务

请参考 创建 Binlog 集群 完成 OceanBase Binlog 服务的安装部署。

Flink 环境设置

下载 Flink 和所需要的依赖包：

1. 通过 下载地址 下载 Flink。本文档使用的是 Flink 1.15.3，并将其解压至目录 flink-1.15.3。

2. 下载下面列出的依赖包，并将它们放到目录 flink-1.15.3/lib/ 下。

   • flink-sql-connector-oceanbase-cdc-2.2.0.jar

   • flink-connector-jdbc-1.15.3.jar

   • mysql-connector-java-5.1.47.jar

准备数据

准备 OceanBase 数据库数据

在 OceanBase 数据库中准备测试数据，作为导入 MySQL 数据库的源数据。

1. 登录 OceanBase 数据库。

   使用 root 用户登录集群的 mysql001 租户。

   [xxx@xxx /home/admin]
   $obclient -h10.10.10.2 -P2881 -uroot@mysql001 -p****** -A
   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_ob_to_mysql，表 tbl1 和 tbl2，并插入数据。

   obclient [(none)]> CREATE DATABASE test_ob_to_mysql;
   Query OK, 1 row affected
   
   obclient [(none)]> USE test_ob_to_mysql;
   Database changed
   obclient [test_ob_to_mysql]> CREATE TABLE tbl1(col1 INT PRIMARY KEY, col2 VARCHAR(20),col3 INT);
   Query OK, 0 rows affected
   
   obclient [test_ob_to_mysql]> 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
   
   obclient [test_ob_to_mysql]> CREATE TABLE tbl2(col1 INT PRIMARY KEY,col2 VARCHAR(20));
   Query OK, 0 rows affected
   
   obclient [test_ob_to_mysql]> 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
   

准备 MySQL 数据库数据

在 MySQL 数据库中创建存放源数据的表。

1. 进入 MySQL 数据库。

   [xxx@xxx /home/admin]
   $mysql -hxxx.xxx.xxx.xxx -P3306 -uroot -p******
   <Omit echo information>
   
   MySQL [(none)]>
   

2. 创建数据库 test_ob_to_mysql 和表 ob_tbl1_and_tbl2。

   MySQL [(none)]> CREATE DATABASE test_ob_to_mysql;
   Query OK, 1 row affected
   
   MySQL [(none)]> USE test_ob_to_mysql;
   Database changed
   MySQL [test_ob_to_mysql]> CREATE TABLE ob_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 目录下。

   [xxx@xxx /FLINK_HOME]
   #cd flink-1.15.3
   

2. 使用下面的命令启动 Flink 集群。

   [xxx@xxx /FLINK_HOME/flink-1.15.3]
   #./bin/start-cluster.sh
   

   启动成功的话，可以在 http://localhost:8081/ 访问到 Flink Web UI，如下所示：

   

   说明

   执行 ./bin/start-cluster.sh 后，如果提示：bash: ./bin/start-cluster.sh: Permission denied。需要把 flink-1.15.3 录下的所有 -rw-rw-r-- 权限的文件的权限都设置为 -rwxrwxrwx 权限。

   示例如下：

   
    [xxx@xxx /.../flink-1.15.3]
    # chmod -R 777 /FLINK_HOME/flink-1.15.3/*
    

3. 使用下面的命令启动 Flink SQL CLI。

   [xxx@xxx /FLINK_HOME/flink-1.15.3]
   #./bin/sql-client.sh
   

   启动成功后，可以看到如下的页面：

   

设置 checkpoint

在 Flink SQL CLI 中开启 checkpoint，每隔 3 秒做一次 checkpoint。

Flink SQL> SET execution.checkpointing.interval = 3s;
[INFO] Session property has been set.

创建 OceanBase CDC 表

在 Flink SQL CLI 中创建 OceanBase 数据库对应的表。对于 OceanBase 数据库中 test_ob_to_mysql 的表 tbl1 和 tbl2 使用 Flink SQL CLI 创建对应的表，用于同步这些底层数据库表的数据。

Flink SQL> CREATE TABLE ob_tbl1 (
    col1 INT PRIMARY KEY,
    col2 VARCHAR(20),
    col3 INT)
    WITH ('connector' = 'oceanbase-cdc',
    'scan.startup.mode' = 'initial',
    'tenant-name' = 'mysql001',
    'username' = 'root@mysql001',
    'password' = '******',
    'database-name' = 'test_ob_to_mysql',
    'table-name' = 'tbl1',
    'hostname' = '10.10.10.2',
    'port' = '2881',
    'rootserver-list' = '10.10.10.2:2882:2881',
    'logproxy.host' = '10.10.10.2',
    'logproxy.port' = '2983');
[INFO] Execute statement succeed.

Flink SQL> CREATE TABLE ob_tbl2 (col1 INT PRIMARY KEY,
    col2 VARCHAR(20))
    WITH ('connector' = 'oceanbase-cdc',
    'scan.startup.mode' = 'initial',
    'tenant-name' = 'mysql001',
    'username' = 'root@mysql001',
    'password' = '******',
    'database-name' = 'test_ob_to_mysql',
    'table-name' = 'tbl2',
    'hostname' = '10.10.10.2',
    'port' = '2881',
    'rootserver-list' = '10.10.10.2:2882:2881',
    'logproxy.host' = '10.10.10.2',
    'logproxy.port' = '2983');
[INFO] Execute statement succeed.

有关 OceanBase CDC WITH 选项的详细信息，请参见 OceanBase CDC Connector。

创建 MySQL CDC 表

在 Flink SQL CLI 中创建 MySQL 数据库对应的表。创建 ob_tbl1_and_tbl2 表，用来将同步的数据写入 MySQL 数据库中。

Flink SQL> CREATE TABLE ob_tbl1_and_tbl2(
    col1 INT PRIMARY KEY,
    col2 INT,col3 VARCHAR(20),
    col4 VARCHAR(20))
    WITH ('connector' = 'jdbc',
    'url' = 'jdbc:mysql://xxx.xxx.xxx.xxx:3306/test_ob_to_mysql',
    'username' = 'root',
    'password' = '******',
    'table-name' = 'ob_tbl1_and_tbl2');
[INFO] Execute statement succeed.

有关 JDBC SQL Connector WITH 选项的详细信息，请参见 JDBC SQL Connector。

在 Flink SQL CLI 中将数据写入 MySQL 数据库中

使用 Flink SQL 将表 tbl1 与表 tbl2 关联，并将关联后的信息写入 MySQL 数据库中。

Flink SQL> INSERT INTO ob_tbl1_and_tbl2 
    SELECT t1.col1,t1.col3,t1.col2,t2.col2 
    FROM ob_tbl1 t1,ob_tbl2 t2 
    WHERE t1.col3=t2.col1;
Flink SQL> INSERT INTO ob_tbl1_and_tbl2
>     SELECT t1.col1,t1.col3,t1.col2,t2.col2
>     FROM ob_tbl1 t1,ob_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: 9cd180a65cb4e2c4d1a5a91465aa38a3

查看关联数据写入 MySQL 数据库情况

登录 MySQL 数据数，在 test_ob_to_mysql 库中查看表 ob_tbl1_and_tbl2 的数据。

MySQL [test_ob_to_mysql]> SELECT * FROM ob_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