使用 RisingWave CDC 接入 OceanBase MySQL 数据

更新时间：2026-04-09 14:12:04

概述

RisingWave 是一款专为云环境设计的分布式 SQL 流处理数据库，致力于降低实时应用的开发成本。本文介绍如何利用 RisingWave 的 MySQL-CDC（Change Data Capture）功能来接入 OceanBase 数据库。通过 CDC 技术，RisingWave 能够实时捕获 OceanBase MySQL 数据库中的数据变更事件，并将这些变更以流的形式进行处理和分析。

版本兼容性

OceanBase 数据库版本：≥ V4.2.1 BP10、V4.2.5 BP1 。

使用限制

在使用 RisingWave CDC 接入 OceanBase 数据时，请注意以下限制：

权限要求：连接 OceanBase 数据库的用户需要拥有 SELECT、REPLICATION SLAVE、REPLICATION CLIENT 权限。
server.id 唯一性：server.id 参数需要确保在数据库中的唯一性，一般指定 1000 以上的随机数字。
数据类型映射：OceanBase 数据类型与 RisingWave 数据类型存在映射关系，部分类型可能需要调整。

前提条件

在使用 RisingWave 之前，确认：

您已完成部署 OceanBase 数据库并且创建了 MySQL 模式用户租户。创建用户租户的详细信息，参见创建租户。
已在创建的 MySQL 兼容模式租户下开通 Binlog 服务。详情请参见 OceanBase Binlog 服务。
已安装 PostgreSQL 命令行客户端工具 psql，并确保其版本与目标数据库兼容，可通过终端执行 psql --version 验证安装状态。
已完成 Docker 部署，Docker 服务正在运行，且当前用户具备执行 docker 命令的权限（可通过 docker info 验证）。

操作步骤

步骤一：获取数据库连接串

联系 OceanBase 数据库部署人员获取连接串，例如：

obclient -h$host -P$port -u$user_name -p$password -D$database_name

参数说明：

$host：连接 IP。ODP 连接用 ODP 地址；直连用 OBServer IP。
$port：连接端口。ODP 默认 2883；直连默认 2881。
$database_name：数据库名称。

注意

连接租户的用户需要拥有该数据库的 CREATE、INSERT、DROP、REPLICATION SLAVE、REPLICATION CLIENT 和 SELECT 权限。更多有关用户权限的信息，请参见 MySQL 模式下的权限分类。
$user_name：连接账户。ODP 格式：用户@租户#集群或集群:租户:用户；直连格式：用户@租户。
$password：账户密码。

更多连接串的信息，请参见通过 OBClient 连接 OceanBase 租户。

示例如下：

obclient -hxxx.xxx.xxx.xxx -P2881 -utest_user001@mysql001 -p****** -Dtest

步骤二：部署 RisingWave

使用 Docker 部署 RisingWave：

sudo docker run -d --pull=always -p 4566:4566 -p 5691:5691 risingwavelabs/risingwave:latest single_node

# 确认 RisingWave 是否启动成功。
sudo netstat -ntlp | grep 4566

该命令会启动 RisingWave 单节点实例，监听端口 4566（RisingWave 服务）和 5691（监控界面）。

步骤三：OceanBase 数据库数据准备

在 OceanBase 数据库中创建用于 CDC 的用户和测试表：

CREATE USER '$rs_name'@'%' IDENTIFIED BY '$rs_password';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '$rs_name'@'%';
FLUSH PRIVILEGES;

create table tbl_t1 (
    c01 int,
    c02 char,
    c03 varchar(10),
    c04 text,
    c05 bigint,
    c06 long,
    c07 numeric(10, 2),
    c08 decimal(20, 8),
    c09 date,
    c10 datetime,
    c11 datetime(3),
    c12 timestamp(6),
    primary key (c01, c02)
);

insert into tbl_t1 (c01, c02, c03, c04, c05, c06, c07, c08, c09, c10, c11, c12)
values (1, 'a', 'abc', 'abcdefg', 123456789000, 99999999999999, 123.456, 12345.6789,
        current_date,
        current_timestamp,
        '2024-01-01 18:18:18.888',
        '2024-01-01 06:06:06.666666');

select * from tbl_t1;

步骤四：RisingWave 连接 OceanBase

使用 psql 连接 RisingWave：

psql -h 127.0.0.1 -p 4566 -d dev -U root

在 RisingWave 中创建 schema 和 MySQL-CDC Source：

-- 创建 schema
create schema rwtest;

-- 创建 MySQL-CDC Source
drop source if exists rwtest.mysql_cdc cascade;
create source rwtest.mysql_cdc
  with (
       connector = 'mysql-cdc',
       hostname = '$host',
       port = '$port',
       username = '$rs_name',
       password = '$rs_password',
       database.name = '$database_name',
       server.id = 1001,       -- 一般指定 1000 以上的随机数字，请确保 server.id 在数据库中的唯一性
       -- 可以指定 debezium 参数配置，如跳过未知的 DDL 语句
       debezium.schema.history.internal.store.only.captured.tables.ddl = 'true'
       );

show sources from rwtest;

-- 创建 CDC 表来映射 OceanBase 表
DROP TABLE IF EXISTS rwtest.test_mysql_cdc;
CREATE TABLE rwtest.test_mysql_cdc (
    c01 int,
    c02 string,          -- 字符串类型不需要定义精度
    c03 string,
    c04 string,
    c05 bigint,
    c06 string,          
    c07 numeric,         -- 数字不需要定义精度
    c08 numeric,
    c09 date,
    c10 timestamp,       -- 时间戳类型不需要定义精度
    c11 timestamp,
    c12 timestamptz,
    PRIMARY KEY (c01, c02)
) FROM rwtest.mysql_cdc TABLE '$database_name.tbl_t1';

select * from rwtest.test_mysql_cdc;

结果验证

查询 CDC 表验证数据同步结果：

dev=> select * from rwtest.test_mysql_cdc;
 c01 | c02 | c03 |   c04   |     c05      |      c06       |  c07   |      c08       |    c09     |         c10         |           c11           |
        c12
-----+-----+-----+---------+--------------+----------------+--------+----------------+------------+---------------------+-------------------------+-------
---------------------------
   1 | a   | abc | abcdefg | 123456789000 | 99999999999999 | 123.46 | 12345.67890000 | 2026-02-05 | 2026-02-05 14:17:17 | 2024-01-01 18:18:18.888 | 2023-1
2-31 22:06:06.666666+00:00
(1 row)