使用 SeaTunnel 读取和同步 OceanBase 数据

更新时间：2024-10-10 21:40:45

SeaTunnel 是一个分布式、高扩展性的数据集成平台，用于在各种数据存储系统之间传输数据。SeaTunnel 提供易于配置的数据流任务插件，让用户定义数据处理和转换的流程。更多关于 SeaTunnel 的信息，参考 SeaTunnel 的 GitHub 地址。

前提条件

在同步数据前，确认以下信息：

您已下载 SeaTunnel。
您已安装 Java 1.8 或更高版本。
您已下载 MySQL 驱动到 lib 目录。
您已安装 oblogproxy。更多信息，参考安装 oblogproxy 文档。

通过 SeaTunnel 读数据

使用 SeaTunnel 同步数据

介绍使用 SeaTunnel 的 MySQL CDC Source Connector 在 OceanBase 数据库 MySQL 租户执行 SQL 查询。connector-cdc-mysql 是 SeaTunnel的一个插件，用于捕获并流式传输 MySQL 数据库中的实时更改数据（Change Data Capture，CDC）。

操作步骤

按照以下步骤进行 OceanBase 数据库 MySQL 租户间的数据同步：

创建 SeaTunnel 用户。

 CREATE USER '用户名'@'%' IDENTIFIED BY '密码';
 GRANT SELECT ON *.* TO '用户名';

配置 config/plugin_config。

指定 connector-cdc-mysql，执行 sh bin/install-plugin.sh 2.3.3。

配置 SeaTunnel。

修改配置文件 config/v2.streaming.conf.template：

source {
# This is a example source plugin **only for test and demonstrate the feature source plugin**
MySQL-CDC {
result_table_name = "fake"
parallelism = 1
server-id = 5656
username = "用户名"
password = "密码"
table-names = ["seatunnel.full_types"]
base-url = "jdbc:mysql://xx.xxx.x.xx:2881/seatunnel"
}

# If you would like to get more information about how to configure SeaTunnel and see full list of source plugins,
# please go to https://seatunnel.apache.org/docs/category/source-v2
}

启动 SeaTunnel。

执行以下命令，启动 SeaTunnel：

./bin/seatunnel.sh --config ./config/v2.streaming.conf.template -e local

介绍使用 SeaTunnel 的 JDBC OceanBase Source Connector 和 JDBC OceanBase Sink Connector 同步数据。JDBC OceanBase Source Connector 支持对 OceanBase 单表数据的全量读和写。更多信息，参考 JDBC OceanBase Source Connector。JDBC OceanBase Sink Connector 支持对 OceanBase 多表数据的全量读和写。更多信息，参考 JDBC OceanBase Sink Connector。

操作步骤

按照以下步骤进行 OceanBase 数据库 MySQL 租户间的数据同步：

创建 SeaTunnel 用户。

 CREATE USER '用户名'@'%' IDENTIFIED BY '密码';
 GRANT SELECT, INSERT, UPDATE, DELETE ON *.* TO '用户名';

配置 config/plugin_config。
下载 OceanBase JDBC 驱动到 lib 目录。更多信息，参考 OceanBase Client。

配置 SeaTunnel。

修改配置文件 config/ob_to_ob：

env {
parallelism = 1
job.mode = "BATCH"
}

source {
# This is a example source plugin **only for test and demonstrate the feature source plugin**
Jdbc {
    driver = com.oceanbase.jdbc.Driver
    url = "jdbc:oceanbase://e2e_oceanbase_mysql:2881/seatunnel?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true&serverTimezone=UTC"
    user = "用户名"
    password = "密码"
    query = "SELECT c_bit_1, c_bit_8 FROM source"
    compatible_mode = "mysql"
}

# If you would like to get more information about how to configure seatunnel and see full list of source plugins,
# please go to https://seatunnel.apache.org/docs/connector-v2/source/FakeSource
}

sink {
Jdbc {
    driver = com.oceanbase.jdbc.Driver
    url = "jdbc:oceanbase://e2e_oceanbase_mysql:2881/seatunnel?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true&serverTimezone=UTC"
    user = "用户名"
    password = "密码"
    query = "insert into sink(c_bit_1, c_bit_8) values (?, ?);"
    compatible_mode = "mysql"
}
# If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
# please go to https://seatunnel.apache.org/docs/connector-v2/sink
}

启动 SeaTunnel。

执行以下命令，启动 SeaTunnel：
```
./bin/seatunnel.sh --config ./config/ob_to_ob -e local
```

注意

OceanBase 数据库不支持 SET 类型。
Caused by: org.apache.seatunnel.api.table.catalog.DataTypeConvertException: ErrorCode:[COMMON-07], ErrorDescription:[Unsupported data type] - Convert type: SET to SeaTunnel data type error.

更多信息，参考 SeaTunnel MySQL CDC 文档。