通过 Apache NiFi 集成 OceanBase 数据库

Apache NiFi 是一个自动化数据流处理平台，支持在系统间实现高效可靠的数据传输。通过读取 OceanBase 数据库的 Binlog 数据，可将数据分发至文件系统、消息队列（如 Kafka）或 HTTP 端点等目标。

前提条件

• 您已完成部署 OceanBase 数据库并且创建了 MySQL 模式用户租户。创建用户租户的详细信息，参见 创建租户。

  • 在创建的 MySQL 兼容模式租户下开通 Binlog 服务。详情请参见 OceanBase Binlog 服务。

• 您的运行环境需满足 Java Development Kit (JDK) 1.8 或更高版本。

• 您已下载 Apache NiFi 安装包。

操作步骤

步骤一：获取数据库连接信息

联系 OceanBase 数据库部署人员或者管理员获取相应的数据库连接串，例如：

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 模式下的权限分类。

• $user_name：提供租户的连接账户。ODP 连接的常用格式：用户名@租户名#集群名 或者 集群名:租户名:用户名；直连方式格式：用户名@租户名。

• $password：提供账户密码。

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

步骤二：配置 NiFi 运行环境

1. 解压 NiFi 安装包至目标目录。

2. 按照如下方式修改 conf/nifi.properties 配置文件（本文以 HTTP 连接为例）：

   # 禁用 HTTPS 并配置 HTTP 访问
   nifi.web.https.host=
   nifi.web.https.port=
   nifi.web.http.host=0.0.0.0
   nifi.web.http.port=<自定义端口号>
   
   # 允许非安全远程连接
   nifi.remote.input.secure=false
   

3. 上传 MySQL 驱动至 NiFi 服务器。

4. 在 NiFi 安装目录下执行如下命令启动服务。如需停止服务，则执行 bin/nifi.sh stop。

   # 启动服务
   bin/nifi.sh start
   

步骤三：配置数据流处理管道

1. 通过浏览器访问 http://<nifi_server_ip>:<port>/nifi/ 进入 NiFi 控制台。

2. 配置 CaptureChangeMySQL Processor。

   1. 拖拽页面顶部的 Processor 图标至画布，搜索并选择 CaptureChangeMySQL，然后单击 ADD。

      

   2. 在 CaptureChangeMySQL Processor 上单击右键选择 Configure，然后在 PROPERTIES 页签下配置相关参数，完成后单击 APPLY。

      属性名称	说明
      MySQL Nodes	OceanBase 数据库地址及端口。
      MySQL Driver Class Name	JDBC 驱动类名，为 com.mysql.cj.jdbc.Driver。
      MySQL Driver Location	您所上传的驱动 JAR 文件所在目录。
      Username	您所创建的数据库用户的用户名。
      Password	数据库用户的密码。
      Database/Schema Name Pattern	需要读取 Binlog 的数据库（支持正则表达式）。
      Table Name Pattern	需要读取 Binlog 的表（支持正则表达式）。
      Include DDL Events	捕获 DDL 变更事件，需设置为 true。

      

3. 配置 PutFile Processor。

   1. 再次拖拽页面顶部的 Processor 图标至画布，搜索并选择 PutFile，然后单击 ADD。

   2. 在 PutFile Processor 上单击右键选择 Configure，然后在 PROPERTIES 页签下填写 Directory 属性，Binlog 文件将会输出到此目录下。完成后单击 APPLY。

      

4. 建立数据流连接。

   1. 回到画布，单击 CaptureChangeMySQL 上的箭头指向 PutFile，将两者进行连接。

   2. 在弹出的 Create Connection 弹窗中，单击 ADD。

   3. 在画布中，再次在 PutFile Processor 上单击右键选择 Configure，在 RELATIONSHIPS 页签下配置 failure 和 success 的行为策略，均选择 terminate。 完成后单击 APPLY。

      

5. 启动数据处理流程。

   1. 回到画布，在 CaptureChangeMySQL Processor 上单击右键选择 Start。

   2. 在 PutFile Processor 上同样单击右键选择 Start。

步骤四：验证数据同步

1. 在 OceanBase 数据库中插入数据。

2. 检查 PutFile 目录下有文件生成。

   

3. 查看文件内容，可以查看执行的 SQL 语句的相关信息。