--- title: "使用 Flink Connector OceanBase 实时同步数据到 OceanBase - OceanBase 数据库 V4.3.5 | OceanBase 文档中心" description: 使用 Flink Connector OceanBase 实时同步数据到 OceanBase Apache Flink 是一个开源的流处理框架，专为分布式、高吞吐和高性能的实时数据处理场景设计。OceanBase 官方提供了基于 JDBC 的 Flink Connector，可以在 Flink 中通过标准 JDBC … --- 切换语言 - 中文站 - 简体中文 - International - English - 日本站 - 日本語 文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 4.3.5 LTS # 使用 Flink Connector OceanBase 实时同步数据到 OceanBase 更新时间：2026-04-14 14:16:46 [编辑](https://github.com/oceanbase/oceanbase-doc/edit/V4.3.5/zh-CN/680.ecological-integration/400.data-ingestion/100.flink/400.synchronize-data-in-real-time-using-flink.md) Apache Flink 是一个开源的流处理框架，专为分布式、高吞吐和高性能的实时数据处理场景设计。OceanBase 官方提供了基于 JDBC 的 Flink Connector，可以在 Flink 中通过标准 JDBC 协议将数据高效写入 OceanBase，支持 **MySQL 兼容模式** 和 **Oracle 兼容模式**，适用于实时数据同步、流式计算结果写入等场景。 ## 连接器特点 **Flink Connector OceanBase（JDBC 连接器）** 通过 JDBC 驱动将数据写入 OceanBase，具有以下核心特点： - **标准 JDBC 协议**：使用成熟的 JDBC 标准协议与 OceanBase 通信，兼容性好、易于集成。 - **双模式支持**：同时支持 OceanBase MySQL 兼容模式和 Oracle 兼容模式。 - **灵活的驱动选择**： - MySQL 兼容模式下可使用 **MySQL JDBC 驱动** 或 **OceanBase JDBC 驱动**。 - Oracle 兼容模式下必须使用 **OceanBase JDBC 驱动**。 - **连接池管理**：基于 Druid 连接池，支持连接复用和参数调优。 - **批量写入优化**：支持缓冲和批量提交，提升写入吞吐量。 **适用场景**：实时数据同步（CDC、流式计算结果写入）。 ## 版本要求与依赖 | 软件 | 版本要求 | 说明 | | --- | --- | --- | | **Apache Flink** | 1.15 或更高版本 | 推荐使用 1.15+ 以获得更好的性能和稳定性 | | **JDK** | 8 或更高版本 | — | | **OceanBase** | 推荐 V3.x、V4.x 及以上 | 支持 MySQL 模式和 Oracle 模式。确认您已安装 OceanBase 数据库并且创建了 MySQL/Oracle 模式租户。 | ## 关于 JDBC 驱动 根据使用的 OceanBase 模式，需要准备相应的 JDBC 驱动： ### MySQL 兼容模式 可选择使用以下任一驱动： - **MySQL JDBC 驱动**（推荐使用 MySQL Connector/J 8.0 或以上版本） - 下载地址：https://dev.mysql.com/downloads/connector/j/ - **OceanBase JDBC 驱动** - 依赖名称：`com.oceanbase:oceanbase-client:2.4.9` - 下载地址：https://mvnrepository.com/artifact/com.oceanbase/oceanbase-client #### 说明 需要将下载的驱动 JAR 文件放入 Flink 的 `lib` 目录下，具体操作步骤，参见后续章节。 ### Oracle 兼容模式（OceanBase 数据库企业版） 必须使用 OceanBase JDBC 驱动： ## 操作步骤 ### 步骤一：快速部署本地 Flink 集群（单机版） #### 下载 Apache Flink 前往 [Flink 官方下载页面](https://flink.apache.org/downloads.html)，选择 **Stable Release**（推荐 1.15+，如 1.18 或 1.19）。 例如（Linux / macOS 终端命令）： ```bash # 下载 Flink（以 1.19.3 为例） wget https://archive.apache.org/dist/flink/flink-1.19.3/flink-1.19.3-bin-scala_2.12.tgz # 解压 tar -xzf flink-1.19.3-bin-scala_2.12.tgz # 进入目录 cd flink-1.19.3 # 设置环境变量（可选但推荐） export FLINK_HOME=$(pwd) ``` #### 启动本地 Flink 集群 在 Flink 目录下执行： ```bash # 启动集群（包含 JobManager + TaskManager） $FLINK_HOME/bin/start-cluster.sh ``` 成功启动后，您会看到类似输出： ```shell Starting cluster. Starting standalonesession daemon on host your-hostname. Starting taskexecutor daemon on host your-hostname. ``` 如果启动失败，可以检查日志： ```shell # 查看 JobManager 日志 tail -f $FLINK_HOME/log/flink-*-standalonesession-*.log # 查看 TaskManager 日志 tail -f $FLINK_HOME/log/flink-*-taskexecutor-*.log ``` #### 验证 Flink 是否正常运行 打开浏览器访问：http://localhost:8081，您可以看到 Flink Dashboard，显示 1 个 TaskManager，可用 slots ≥ 1。 ### 步骤二：下载并部署 JDBC 连接器 1. 下载基于 JDBC 的连接器 JAR 包 `flink-sql-connector-oceanbase-${version}.jar`： - 下载地址：https://repo1.maven.org/maven2/com/oceanbase/flink-sql-connector-oceanbase/ - 选择版本（例如：1.6.0），下载对应的 JAR 文件 #### 注意 文件名必须是 `flink-sql-connector-oceanbase-.jar`，这是 Flink SQL 识别连接器的关键。 2. 将 JAR 文件复制到 Flink 的 `lib` 目录： ```bash # 假设下载的文件在当前目录，FLINK_HOME 是 Flink 安装目录 cp flink-sql-connector-oceanbase-1.6.0.jar $FLINK_HOME/lib/ ``` ### 步骤三：准备并部署 JDBC 驱动 根据您使用的 OceanBase 模式，下载并放置相应的 JDBC 驱动： #### 方式一：MySQL 兼容模式 可选择使用以下任一驱动： 1. **MySQL JDBC 驱动**（推荐 8.0 或以上版本） - 下载地址：https://dev.mysql.com/downloads/connector/j/ - 选择 Platform Independent 版本（ZIP 或 TAR.GZ） - 解压后找到 `mysql-connector-j-x.x.x.jar` 文件 2. **OceanBase JDBC 驱动** - 下载地址：https://mvnrepository.com/artifact/com.oceanbase/oceanbase-client - 选择最新版本下载 JAR 文件 将下载的驱动 JAR 文件放入 Flink 的 `lib` 目录： ```bash # 示例：复制 MySQL JDBC 驱动 cp mysql-connector-j-8.0.33.jar $FLINK_HOME/lib/ ``` #### 方式二：Oracle 兼容模式 必须使用 OceanBase JDBC 驱动： - 下载地址：https://mvnrepository.com/artifact/com.oceanbase/oceanbase-client - 选择最新版本（例如：2.4.9）下载 JAR 文件 ```bash # 复制 OceanBase JDBC 驱动 cp oceanbase-client-2.4.9.jar $FLINK_HOME/lib/ ``` #### 重启 Flink 集群（使 JAR 生效） ```bash # 先停止 $FLINK_HOME/bin/stop-cluster.sh # 再启动 $FLINK_HOME/bin/start-cluster.sh ``` ### 步骤四：获取数据库连接信息 联系 OceanBase 数据库部署人员或管理员获取数据库连接信息，并登录。例如： ```shell obclient -h$host -P$port -u$user_name -p$password -D$database_name ``` **参数说明：** - `$host`：OceanBase 数据库连接 IP。OceanBase 数据库代理（OceanBase Database Proxy，ODP）连接方式使用 ODP 地址；直连方式使用 OBServer 节点的 IP 地址。 - `$port`：OceanBase 数据库连接端口。ODP 连接方式默认端口为 `2883`，在部署 ODP 时可自定义；直连方式默认端口为 `2881`，在部署 OceanBase 数据库时可自定义。 - `$database_name`：需要访问的数据库名称。 #### 注意 连接租户的用户需要拥有该数据库的 `CREATE`、`INSERT`、`DROP` 和 `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)。 #### 确认网络连通性 确保运行 Flink 作业的环境能够访问 OceanBase 数据库： 测试命令： ```bash # 测试 RPC 端口连通性 nc -zv ``` - ：也就是连接串中的 `$host`，OceanBase 数据库连接 IP。OceanBase 数据库代理（OceanBase Database Proxy，ODP）连接方式使用 ODP 地址；直连方式使用 OBServer 节点的 IP 地址。 - ：也就是连接串中的 `$port`：OceanBase 数据库连接端口。ODP 连接方式默认端口为 `2883`，在部署 ODP 时可自定义；直连方式默认端口为 `2881`，在部署 OceanBase 数据库时可自定义。 ### 步骤五：在 OceanBase 中准备目标表 在 OceanBase 数据库 MySQL 兼容模式下的 `test` 库中创建目标表 `t_sink`： ```sql USE test; CREATE TABLE `t_sink` ( `id` int(10) NOT NULL, `username` varchar(20) DEFAULT NULL, `score` int(10) DEFAULT NULL, PRIMARY KEY (`id`) ); ``` ### 步骤六：启动 Flink SQL Client 并测试 使用 SQL Client 执行 SQL 语句并写入数据： ```bash # 进入 Flink 安装目录（如果还未在该目录） cd $FLINK_HOME # 启动 SQL Client ./bin/sql-client.sh ``` 成功启动后，您将看到 Flink SQL Client 的交互式命令行界面： ```shell ▒▓██▓██▒ ▓████▒▒█▓▒▓███▓▒ ▓███▓░░ ▒▒▒▓██▒ ▒ ░██▒ ▒▒▓▓█▓▓▒░ ▒████ ██▒ ░▒▓███▒ ▒█▒█▒ ░▓█ ███ ▓░▒██ ▓█ ▒▒▒▒▒▓██▓░▒░▓▓█ █░ █ ▒▒░ ███▓▓█ ▒█▒▒▒ ████░ ▒▓█▓ ██▒▒▒ ▓███▒ ░▒█▓▓██ ▓█▒ ▓█▒▓██▓ ░█░ ▓░▒▓████▒ ██ ▒█ █▓░▒█▒░▒█▒ ███▓░██▓ ▓█ █ █▓ ▒▓█▓▓█▒ ░██▓ ░█░ █ █▒ ▒█████▓▒ ██▓░▒ ███░ ░ █░ ▓ ░█ █████▒░░ ░█░▓ ▓░ ██▓█ ▒▒▓▒ ▓███████▓░ ▒█▒ ▒▓ ▓██▓ ▒██▓ ▓█ █▓█ ░▒█████▓▓▒░ ██▒▒ █ ▒ ▓█▒ ▓█▓ ▓█ ██▓ ░▓▓▓▓▓▓▓▒ ▒██▓ ░█▒ ▓█ █ ▓███▓▒░ ░▓▓▓███▓ ░▒░ ▓█ ██▓ ██▒ ░▒▓▓███▓▓▓▓▓██████▓▒ ▓███ █ ▓███▒ ███ ░▓▓▒░░ ░▓████▓░ ░▒▓▒ █▓ █▓▒▒▓▓██ ░▒▒░░░▒▒▒▒▓██▓░ █▓ ██ ▓░▒█ ▓▓▓▓▒░░ ▒█▓ ▒▓▓██▓ ▓▒ ▒▒▓ ▓█▓ ▓▒█ █▓░ ░▒▓▓██▒ ░▓█▒ ▒▒▒░▒▒▓█████▒ ██░ ▓█▒█▒ ▒▓▓▒ ▓█ █░ ░░░░ ░█▒ ▓█ ▒█▓ ░ █░ ▒█ █▓ █▓ ██ █░ ▓▓ ▒█▓▓▓▒█░ █▓ ░▓██░ ▓▒ ▓█▓▒░░░▒▓█░ ▒█ ██ ▓█▓░ ▒ ░▒█▒██▒ ▓▓ ▓█▒ ▒█▓▒░ ▒▒ █▒█▓▒▒░░▒██ ░██▒ ▒▓▓▒ ▓██▓▒█▒ ░▓▓▓▓▒█▓ ░▓██▒ ▓░ ▒█▓█ ░░▒▒▒ ▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓ ▓░▒█░ ______ _ _ _ _____ ____ _ _____ _ _ _ BETA | ____| (_) | | / ____|/ __ \| | / ____| (_) | | | |__ | |_ _ __ | | __ | (___ | | | | | | | | |_ ___ _ __ | |_ | __| | | | '_ \| |/ / \___ \| | | | | | | | | |/ _ \ '_ \| __| | | | | | | | | < ____) | |__| | |____ | |____| | | __/ | | | |_ |_| |_|_|_| |_|_|\_\ |_____/ \___\_\______| \_____|_|_|\___|_| |_|\__| Welcome! Enter 'HELP;' to list all available commands. 'QUIT;' to exit. Flink SQL> ``` 现在您可以在 SQL Client 中执行 SQL 语句了。 ### 步骤七：配置 JDBC 连接并写入数据 #### MySQL 兼容模式 在 SQL Client 中创建 Flink 目标表，映射到 OceanBase 的 `t_sink` 表： ```sql CREATE TABLE t_sink ( id INT, username VARCHAR, score INT, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'oceanbase', 'url' = 'jdbc:mysql://127.0.0.1:2881/test', ------仅为示例，请根据实际环境填写正确的 url 'schema-name' = 'test', 'table-name' = 't_sink', 'username' = 'root@test#obcluster', ------仅为示例，请根据连接串中获取到的实际租户名填写 'password' = 'your_password', ------请根据连接串中获取到的实际密码填写 'druid-properties' = 'druid.initialSize=10;druid.maxActive=100;', 'buffer-flush.interval' = '1s', 'buffer-flush.buffer-size' = '5000', 'max-retries' = '3' ); ``` **配置说明**： - `url`：OceanBase 的 JDBC 连接地址，格式为 `jdbc:mysql://host:port/database` - `schema-name` 和 `table-name`：目标数据库和表名 - `username` 和 `password`：OceanBase 数据库用户名和密码 - 其他参数用于调优性能和可靠性 创建表成功后，插入测试数据： ```sql INSERT INTO t_sink VALUES (1, 'Tom', 99), (2, 'Jerry', 88), (1, 'Tom', 89); ``` 执行成功后，您将看到类似如下的输出： ```shell [INFO] Submitting SQL update statement to the cluster... [INFO] SQL update statement has been successfully submitted to the cluster: Job ID: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx ``` **验证数据写入** 在 OceanBase 数据库中查询验证数据是否成功写入： ```bash # 连接到 OceanBase MySQL 兼容模式，以下仅为示例，请根据连接串获取到的连接信息正确填写 mysql -h127.0.0.1 -P2881 -uroot@test#obcluster -p ``` ```sql -- 查询数据 USE test; SELECT * FROM t_sink; ``` **预期结果**： ```shell +----+----------+-------+ | id | username | score | +----+----------+-------+ | 1 | Tom | 89 | | 2 | Jerry | 88 | +----+----------+-------+ 2 rows in set ``` #### 说明 - 由于主键冲突，`id=1` 的记录会被更新为最新的值（Tom, 89）。 - 连接器通过 JDBC 协议与 OceanBase 通信，使用标准的 UPSERT 语义。 #### Oracle 兼容模式 对于 OceanBase 数据库企业版的 Oracle 兼容模式，需要指定 OceanBase JDBC 驱动对应的 `url` 和 `driver-class-name`。 在 SQL Client 中执行： ```sql CREATE TABLE t_sink ( id INT, username VARCHAR, score INT, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'oceanbase', 'url' = 'jdbc:oceanbase://127.0.0.1:2881/SYS', ------仅为示例，请根据实际环境填写正确的 url 'driver-class-name' = 'com.oceanbase.jdbc.Driver', 'schema-name' = 'SYS', 'table-name' = 'T_SINK', 'username' = 'SYS@test#obcluster', ------仅为示例，请根据连接串中获取到的实际租户名填写 'password' = 'pswd', ------请根据连接串中获取到的实际密码填写 'druid-properties' = 'druid.initialSize=10;druid.maxActive=100;', 'buffer-flush.interval' = '1s', 'buffer-flush.buffer-size' = '5000', 'max-retries' = '3' ); ``` **Oracle 模式配置要点**： - `url`：使用 `jdbc:oceanbase://` 协议 - `driver-class-name`：必须指定为 `com.oceanbase.jdbc.Driver` - `schema-name` 和 `table-name`：Oracle 模式下通常使用大写 插入测试数据： ``` ```bash # 连接到 OceanBase Oracle 兼容模式，以下仅为示例，请根据连接串获取到的连接信息正确填写 obclient -h127.0.0.1 -P2881 -uSYS@test#obcluster -p ``` ```sql -- 查询数据 SELECT * FROM T_SINK; ``` #### 说明 Oracle 模式同样支持 UPSERT 语义，主键冲突时会自动更新为最新值。 ### 步骤八：关闭和清理（可选） 完成测试后，您可以停止 Flink 集群： ```bash # 停止 Flink 集群 cd $FLINK_HOME ./bin/stop-cluster.sh ``` 在 SQL Client 中退出： ```sql -- 在 SQL Client 中执行 QUIT; ``` ## JDBC 连接器配置说明 以下是 JDBC 连接器的主要配置项： | 参数名 | 必需 | 默认值 | 类型 | 描述 | | --- | --- | --- | --- | --- | | `url` | 是 | - | String | 数据库的 JDBC url | | `username` | 是 | - | String | 连接用户名 | | `password` | 是 | - | String | 连接密码 | | `schema-name` | 是 | - | String | 连接的 schema 名或 db 名 | | `table-name` | 是 | - | String | 表名 | | `driver-class-name` | 否 | `com.mysql.cj.jdbc.Driver` | String | JDBC 驱动类名。 MySQL 兼容模式可使用 `'com.mysql.cj.jdbc.Driver'`（默认值）或 `'com.oceanbase.jdbc.Driver'`， Oracle 兼容模式必须使用 `'com.oceanbase.jdbc.Driver'` | | `druid-properties` | 否 | - | String | Druid 连接池属性，多个值用分号分隔 | | `sync-write` | 否 | `false` | Boolean | 是否开启同步写，设置为 `true` 时将不使用 buffer 直接写入数据库 | | `buffer-flush.interval` | 否 | `1s` | Duration | 缓冲区刷新周期。设置为 `'0'` 时将关闭定期刷新 | | `buffer-flush.buffer-size` | 否 | `1000` | Integer | 缓冲区大小 | | `max-retries` | 否 | `3` | Integer | 失败重试次数 | | `memstore-check.enabled` | 否 | `true` | Boolean | 是否开启内存检查 | | `memstore-check.threshold` | 否 | `0.9` | Double | 内存使用的阈值相对最大限制值的比例 | | `memstore-check.interval` | 否 | `30s` | Duration | 内存使用检查周期 | | `partition.enabled` | 否 | `false` | Boolean | 是否启用分区计算功能，按照分区来写数据。 仅当 `'sync-write'` 和 `'direct-load.enabled'` 都为 `false` 时生效 | #### 说明 该连接器基于 JDBC 协议，通过 Druid 连接池管理数据库连接。 ## 使用场景 ### 实时数据同步 基于 JDBC 连接器进行实时数据写入，利用异步批量写入的优势提升吞吐量。JDBC 连接器通过内置的缓冲机制，将多条记录批量提交到 OceanBase，适合持续的实时数据流写入。 #### 典型应用场景： 1. **与 CDC 结合的数据库同步** JDBC 连接器可以与 CDC 数据源（如 MySQL CDC）结合使用，进行数据库之间的实时同步。JDBC 连接器作为 Sink 端，通过标准 JDBC 协议将 CDC 捕获的变更数据写入 OceanBase。 2. **流式计算结果写入** 将 Flink 流式计算的结果实时写入 OceanBase，支持复杂的数据处理和转换。 3. **实时业务数据同步** 持续同步业务系统的数据变更到 OceanBase，用于数据仓库实时更新、实时报表等场景。 ## 使用 Flink Connector 的最佳实践 ### 调整并行度提升性能 通过调整 Flink 任务的并行度可以显著提升写入性能。并行度越高，同时写入 OceanBase 的连接数越多，吞吐量越大。 **设置方式**：在 Flink SQL Client 中设置全局并行度： ```sql SET 'parallelism.default' = '8'; ``` **并行度调优建议**： - 根据数据量和 OceanBase 集群性能合理设置并行度 - 监控 OceanBase 的负载情况，避免写入压力过大 ### 批量写入性能优化 调整 `buffer-flush.buffer-size` 和 `buffer-flush.interval` 参数以平衡延迟和吞吐量： - `buffer-flush.buffer-size`：缓冲区大小，默认 1000 条 - `buffer-flush.interval`：缓冲区刷新周期，默认 1s **高吞吐场景**可适当增大 `buffer-size` 和延长 `interval`。 ### 可靠性保障 - 配置 `max-retries` 参数控制失败重试次数（默认 3 次） - 启用 `memstore-check.enabled` 避免 OceanBase 内存溢出（默认开启） - 使用 `sync-write=true` 可确保数据实时可见性，但会牺牲性能（每条记录直接通过 JDBC 提交） ### 监控建议 - 监控 Flink 任务的背压情况，及时发现性能瓶颈 - 关注 OceanBase 的写入性能指标（TPS、QPS、响应时间等） - 监控各并行任务的数据倾斜情况，确保负载均衡 ## 相关文档 - [Flink 项目 GitHub 仓库](https://github.com/oceanbase/flink-connector-oceanbase) - [Flink 连接器 GitHub 仓库](https://repo1.maven.org/maven2/com/oceanbase/flink-sql-connector-oceanbase/) - [社区支持](https://ask.oceanbase.com) 上一篇 下一篇 ![有帮助](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) 咨询热线