---
title: "使用 RisingWave CDC 接入 OceanBase MySQL 数据 - OceanBase 数据库 V4.3.5 | OceanBase 文档中心"
description: 使用 RisingWave CDC 接入 OceanBase MySQL 数据 概述 RisingWave 是一款专为云环境设计的分布式 SQL 流处理数据库，致力于降低实时应用的开发成本。 本文介绍如何利用 RisingWave 的 MySQL-CDC（Change Data Capture）功能来接入 Ocean…
---
切换语言

- 中文站 - 简体中文
- International - English
- 日本站 - 日本語

文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 4.3.5 LTS

# 使用 RisingWave CDC 接入 OceanBase MySQL 数据

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

[编辑](https://github.com/oceanbase/oceanbase-doc/edit/V4.3.5/zh-CN/680.ecological-integration/400.data-ingestion/2100.risingwave.md)  

## 概述

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 模式用户租户。创建用户租户的详细信息，参见 [创建租户](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000002013128)。
 - 已在创建的 MySQL 兼容模式租户下开通 Binlog 服务。详情请参见 [OceanBase Binlog 服务](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000002013297)。
 - 已安装 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 模式下的权限分类](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000002016100)。
 - `$user_name`：连接账户。ODP 格式：`用户@租户#集群`或`集群:租户:用户`；直连格式：`用户@租户`。
 - `$password`：账户密码。

更多连接串的信息，请参见[通过 OBClient 连接 OceanBase 租户](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000002013251)。

**示例如下：**

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

```

### 步骤二：部署 RisingWave

使用 Docker 部署 RisingWave：

```powershell
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 的用户和测试表：

```sql
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：

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

```

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

```sql
-- 创建 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 表验证数据同步结果：

```sql
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)

```

## 相关文档

更多使用方法请参考 [RisingWave 官方文档](https://docs.risingwave.com/ingestion/sources/mysql/mysql-cdc#examples)。

 上一篇 下一篇 ![有帮助](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) 咨询热线