You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

15 KiB

Raw Blame History Unescape Escape

OceanBase CDC 连接器

OceanBase CDC 连接器允许从 OceanBase 读取快照数据和增量数据。本文介绍了如何设置 OceanBase CDC 连接器以对 OceanBase 进行 SQL 查询。

依赖

为了使用 OceanBase CDC 连接器，您必须提供相关的依赖信息。以下依赖信息适用于使用自动构建工具（如 Maven 或 SBT）构建的项目和带有 SQL JAR 包的 SQL 客户端。

<dependency>
  <groupId>com.ververica</groupId>
  <artifactId>flink-connector-oceanbase-cdc</artifactId>
  <!--  请使用已发布的版本依赖，snapshot 版本的依赖需要本地自行编译。 -->
  <version>2.3.0</version>
</dependency>

下载 SQL 客户端 JAR 包

下载链接仅在已发布版本可用，请在文档网站左下角选择浏览已发布的版本。

下载 flink-sql-connector-oceanbase-cdc-2.3.0.jar 到 <FLINK_HOME>/lib/ 目录下。

注意: flink-sql-connector-oceanbase-cdc-XXX-SNAPSHOT 版本是开发分支release-XXX对应的快照版本，快照版本用户需要下载源代码并编译相应的 jar。用户应使用已经发布的版本，例如 flink-sql-connector-oceanbase-cdc-2.3.0.jar 当前已发布的所有版本都可以在 Maven 中央仓库获取。

配置 OceanBase 数据库和 oblogproxy 服务

按照部署文档配置 OceanBase 集群。

在 sys 租户中，为 oblogproxy 创建一个带密码的用户。更多信息，参考用户管理文档。

mysql -h${host} -P${port} -uroot
mysql> SHOW TENANT;
mysql> CREATE USER ${sys_username} IDENTIFIED BY '${sys_password}';
mysql> GRANT ALL PRIVILEGES ON *.* TO ${sys_username} WITH GRANT OPTION;

为你想要监控的租户创建一个用户，这个用户用来读取快照数据和变化事件数据。
OceanBase 社区版用户需要获取rootserver-list，可以使用以下命令获取：
```
mysql> SHOW PARAMETERS LIKE 'rootservice_list';
```
OceanBase 企业版用户需要获取 config-url，可以使用以下命令获取：
```
mysql> show parameters like 'obconfig_url';
```
按照快速入门文档配置 oblogproxy。

创建 OceanBase CDC 表

使用以下命令，创建 OceanBase CDC 表：

-- 每 3 秒做一次 checkpoint，用于测试，生产配置建议 5 到 10 分钟                  
Flink SQL> SET 'execution.checkpointing.interval' = '3s';

-- 在 Flink SQL 中创建 OceanBase 表 `orders`
Flink SQL> CREATE TABLE orders (
    order_id     INT,
    order_date   TIMESTAMP(0),
    customer_name STRING,
    price        DECIMAL(10, 5),
    product_id   INT,
    order_status BOOLEAN,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'oceanbase-cdc',
    'scan.startup.mode' = 'initial',
    'username' = 'user@test_tenant',
    'password' = 'pswd',
    'tenant-name' = 'test_tenant',
    'database-name' = 'test_db',
    'table-name' = 'orders',
    'hostname' = '127.0.0.1',
    'port' = '2881',
    'rootserver-list' = '127.0.0.1:2882:2881',
    'logproxy.host' = '127.0.0.1',
    'logproxy.port' = '2983',
    'working-mode' = 'memory'
);

-- 从表 orders 中读取快照数据和 binlog 数据
Flink SQL> SELECT * FROM orders;

您也可以访问 Flink CDC 官网文档，快速体验将数据从 OceanBase 导入到 Elasticsearch。更多信息，参考 Flink CDC 官网文档。