ChunJun&OceanBase联合方案首次发布：构建一体化数据集成方案

8 月 27 日，ChunJun 社区与 OceanBase 社区联合组织的开源线下 Meetup 成功举办，会上重磅发布了「OceanBase&ChunJun：构建一体化数据集成方案」。

这是 OceanBase&ChunJun 联合解决方案的首次发布，将针对分库分表的实时数据集成、跨集群 / 租户的数据集成、不同数据源的实时数据集成、日志类型数据的全增量一体化处理等诸多场景，提供高可靠数据集成解决方案。

下面为大家带来具体介绍，欢迎分享给更多的开发者和爱好者共同学习、探讨。

课件获取：

关注公众号 “ChunJun”，后台私信 “Meetup” 获得分享课件

视频回看：

https://www.bilibili.com/video/BV1mG41137ZV?spm_id_from=333.999.0.0

ChunJun&OceanBase 是什么

ChunJun：一款稳定、高效、易用的数据集成框架

ChunJun 是一款高效、稳定、易用的数据集成框架，目前基于 Apache Flink 实时计算引擎实现批流一体的数据读取和写入。

file

ChunJun 的核心能力

・多数据源：目前已支持 30 + 数据源，涵盖了各类数据库、文件系统等

・灵活的任务运行模式：支持开箱即用的 local 模式运行，也支持 flink standalone、yarn、k8s 等模式；支持 Taier、DolphinScheduler、Dlinky 等大数据调度平台

・数据还原：支持 DML 和 DDL 同步，可以最大程度保证源端和目标端的数据和结构统一

・断点续传：依托 Flink 的 Checkpoint 机制，可以从失败的位点重试

・速率控制：支持多种分片方式，用户可根据自身业务调整分片逻辑；支持调整读取和写入的并发度，控制每秒读取的数据量

・脏数据管理：支持多种方式存储脏数据，控制脏数据生命周期，并提供统计数据

OceanBase：企业级开源分布式 HTAP 数据库

企业级开源分布式 HTAP（Hybrid Transaction/Analytical Processing）数据库，具有原生分布式架构，支持金融级高可用、透明水平扩展、分布式事务、多租户和语法兼容等企业级特性。

OceanBase 的核心能力

・高可用：基于 Paxos 协议，强一致性；少数副本故障，数据不丢，服务不停；RPO=0; RTO<30s

・高扩展：在线进行水平扩、缩容；自动实现负载均衡

・低成本：不依赖高端硬件，降低成本；极致的压缩比，节省成本

・HTAP：一套计算引擎同时支持混合负载；一套数据库，读写分离

・高兼容：兼容 MySQL 协议与语法；降低业务改造迁移成本

・多租户：一套环境独立运行多套业务；保证租户数据安全

ChunJun OceanBase Connector 实现

OceanBase CDC

OceanBase 作为分布式数据库，日志信息分布在集群当中不同的机器上，需要有一个工具把这些日志信息进行汇总，拿到正确、完整的日志信息。

OceanBase 社区版利用 CDC 组件架构进行这项工作，它主要是通过 oblogproxy 来提供日志拉取的服务，如果想集成 OceanBase 增量数据的处理，可以在自己的业务应用中去集成 oblogclient 来进行处理，目前已对接了 ChunJun、Flink CDC、Cloud Canal 等数据集成框架。

file

OceanBase 社区版 CDC 组件架构

ChunJun Connectors 的工作模式

ChunJun 中的读取和写入主要是通过 Connector 中的一些结构和模块来实现的，包含 RDB、CDC 、NoSQL、MQ、File 等。

・RDB Connectors：基于 JDBC Connector，通过轮询支持了源表包含自增列且增量数据只有 insert 操作时的全增量一体化读取及写入。

・CDC Connectors：基于数据库的 Binlog 或 Redolog，实现增量数据的读取。

file

Flink 流数据与动态表

ChunJun 上的这些数据最终会在 Flink 进行处理，在 Flink 当中通过定义动态表的结构，可以将流数据在执行 SQL 前先转换为可以操作的表，然后通过连续查询来获取一个不断更新的执行结果。

下图就是数据从数据流转成动态表，在流数据上定义一张标，通过执行连续查询来获取不断更新的结果。

file

ChunJun OceanBase Connector 的实现

在 ChunJun 中主要是通过 Chunjun Core 模块来满足将数据读取到 Flink 及从 Flink 中写出去，其中 DynamicTableSourceFactory 及 DynamicTableSinkFactory 支持 SQL 类型的任务，SourceFactory 及 SinkFactory 用来支持 Json 类型的任务。

如下图所示，ChunJun OceanBase Connector 的实现主要通过两种方式：一种是从 Chunjun Core 到 JDBC Connector 再到 OceanBase Connector；另外一种是从 Chunjun Core 直接到 OceanBase CDC Connector。

file

ChunJun & OceanBase 应用

场景 1：针对分库分表的实时数据集成

使用 Oceanbase CDC Connector，库表名利用 Fnmatch 通配，实现分库分表数据源的实时数据集成。这个场景可以做增量同步，也可以做单数据流的 ETL 操作。

file

场景 2：跨集群 / 租户的数据集成

目前，不同租户的数据在一个连接当中获取不到，如果想对 OB 当中不同租户的数据做一个统一处理，需通过多个数据库的连接来实现分别读取，这时可以利用 ChunJun 中与 OceanBase 相关的 connector，读取不同集群、租户数据到 Flink。

file

场景 3：不同数据源的实时数据集成

可以对不同种类数据源进行数据汇聚，使用不同类型数据库的 connector，读取不同数据源的数据到 Flink。

file

场景 4：日志类型数据的全增量一体化处理

对于只有 insert 增量变动的数据源，基于自增列进行全增量一体化的处理。

file

ChunJun&OceanBase 未来展望

● 提高代码质量

・增加测试 case，覆盖所有的启动方式和常见的业务场景

・完全适配 MySQL 5.1.4x 和 8.0 驱动

● 20 + 种丰富的任务类型

・增加非 transformer 模式 sync 任务的支持

・增加 OceanBase 企业版 Oracle 模式的支持

● 提高方案可靠性

・增加数据读取的事务性支持

・简化 oblogproxy 的部署，支持 Docker 部署

・增加详细的使用文档