ChunJun 支持异构数据源 DDL 转换与自动执行 丨 DTMO 02 期回顾（内含课程回放 + 课件）

导读：

4 月 26 日晚，ChunJun 项目核心成员、袋鼠云数栈大数据引擎开发专家渡劫为大家带来分享《ChunJun 支持异构数据源 DDL 转换与自动执行》，我们将直播精华部分做了整理，带大家再次回顾内容，加深技术细节的了解。

你能看到

▫ 数据还原介绍

▫ DDL 自动转换架构设计

▫ Calcite 解析 DDL 实战

直播课件获取：

关注公众号“数栈研习社”，后台私信“ChunJun01”获得直播课件

直播视频回看：

点击“阅读原文”，观看精彩视频

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

演讲 / 渡劫

整理 / 花夏

数据还原介绍

ChunJun 实时同步支持 mysql oracle postgresql sqlserver 等数据源实时同步，但是同步之后的数据是以日志形式输出，数据还原在此基础上做到源数据的变动在目标表也发生对应变动，包含 DML 以及 DDL 的操作都会在目标表中执行对应的操作，保证源表和目标表 schema 一致 数据一致。

目前 ChunJun 数据还原已经支持 mysql 到 rdb 类型数据源的数据还原，仅限于支持 DML 的还原，DDL 的自动执行下一版本支持。

实时还原增加了两个主要模块：

• 源表和目标表的映射(database table column 信息的映射)

• 与外部交互，完成 DDL 状态更新，DML 数据重新下发

为了完成逻辑解耦，我们增加了 2 个 flatMap 算子完成上述操作：

• NameMappingFlatMap 根据映射关系对数据信息进行对应替换

• RestorationFlatMap 对数据进行处理，对数据进行阻塞下发以及 DDL 状态监听

flatMap 算子介绍

接下来为大家介绍两个算子

01

NameMappingFlatMap

实时还原默认 source 端 schema table column 是和 sink 端一致的，但是在大多数情况下 source 和 sink 的映射并不是完全一致的，因此需要 NameMappingFlatMap 算子对 source 的 schema table column 进行替换。NameMapping 支持 schema table column 的映射，其映射关系如下图所示：

图中映射关系代表源表 schema 为 ChunJun_source 下的 source1 这个表对应对应于目标端 ChunJun_sink 下的 sink1，其中字段映射为 源表的 C1 字段对应目标 id 字段，C2 字段对应目标 name 字段

在创建 flink 同步任务的时候，会判断脚本里是否配置了 nameMapping 的配置，如果没有配置则不会存在 NameMappingFlatMap 算子。

02

RestorationFlatMap

在数据还原中一定会涉及到 DDL，但是目前 sink 端只支持 DML 的执行，因此在源表产生 DDL 之后的 DML 数据不能直接发给 sink 端执行，需要等到 sink 端对应的 DDL 执行完之后，DML 才能重新下发。

因此 RestorationFlatMap 设计主要是为了解决 数据的下发 何时下发问题，何时下发就是下游 sink 的 DDL 执行完，但是这个 sink 端 ddl 的执行不是 ChunJun 完成的，ChunJun 是无法得知完成时间的。因此 RestorationFlatMap 会和外部交互 获取这个 DDL 执行状态 从而判断 DML 数据何时下发。

结构设计

RestorationFlatMap 内部会对每个表维护一个集合，DML&DDL 数据都会存入此集合。集合会在非阻塞和阻塞状态间进行切换，同时内部会有两个组件分别为 workerManager 以及 Monitor 组件：

• WorkerManager：监听非阻塞集合数据，如果是 DML 下发，如果是 DDL 则将队列置为阻塞状态

• Monitor：将 ddl 存储到外部数据源 以及监听阻塞队列的 ddl 执行情况，进行阻塞到非阻塞的改变 store 监听阻塞状态队列的第一个 ddl 数据，将其存储到外部表 fetcher 监听外部表 DDL 数据的状态 如果为已执行，则将此表对应的集合阻塞状态改为非阻塞

外部表设计

ChunJun 目前支持 DDL 数据存储外部 Mysql 数据源在数据还原中会将 DDL 数据写入到外部数据源，第三方修改此 DDL 数据的 status 为 2 后，ChunJun 会认为下游 DDL 已执行完毕

数据还原脚本示例

脚本示例主要分为 nameMapping 及 restoration 两部分，分别对应 NameMappingFlatMap 及 RestorationFlatMap 算子参数。

注意 reader 的 split 需要设置为 true。

数据还原存在问题

• binlog 支持 DDL 读取，logminer 暂未支持 需要所有数据源支持 DDL 的读取

• RestorationFlatMap 会将数据存储在内存中，后续会进行外部持久化

• checkpoint 支持不足，RestorationFlatMap 模块数据续跑后会丢失

• 当前数据源产生 DDL 场景和外部交互过多，后续增加 DDL 自动执行，达到 DML&DDL 都由 chunjun 完成，用户无感知

DDL 自动转换架构设计

当前数据还原 DDL 执行依赖外部进行操作，ChunJun 仅根据 DDL 数据状态进行 DML 数据下发，为了做到整条链路由 ChunJun 闭环完成，减少外部依赖以及运维成本，ChunJun 进行 DDL 自动转换操作，将 source 的 DDL 语法转换为 sink 的 DDL 语法，因此就有了 DDL 自动转换模块的设计。

DDL 自动转换解决下列问题：

• 当前 ddl 数据 ChunJun 下游不会自动执行

• 外部表存储的 DDL 数据状态是客户手动修改

主要结构设计：

将 DDL 自动转换逻辑放在 NameMappingFlatMap 中，NameMappingFlatMap 执行数据转换。

DDL 技术方案

数据还原自动转换功能主要是以下三部分：

1、解析 DDL 语句

源表 DDL SQL 转为一个中间对象以及中间对象转为目标端 DDL 语句。

2、异常数据管理

如果自动转换时失败，抛出 conventException 后，由对应的异常管理器处理。

3、DDL 数据状态自动修改

DDL SQL 在下游执行完毕后，基于事件通知方式将中间表存储的 DDL 状态改为已完成。

DDL 架构设计

由于 DDL 没有统一标准，每个数据源的 DDL 语法不同，因此需要按照每个数据源的 DDL 语法进行解析，并将其解析为一个中间数据，然后将这个中间数据转为目标类型数据源的 DDL 语句。

因此 DDLConvent 顶层接口会抽象出三个基本方法：

1、RowData 转为中间数据

2、中间数据转为 DdlSql

3、获取数据源类型

Calcite 解析 DDL 实战

Calcite 解析 DDL 实战基于代码层面做此次演示，就不在公众号上做具体演示回顾，各位社区小伙伴们可前往 B 站查看视频直播回顾。

B 站直播回顾地址：

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

结语

以上就是我们在数据还原上增加的 DDL 自动执行设计思路，我们规划将在上半年完成以上功能点，如果大家有好的想法也欢迎给我们提 issue 或者 pr。

issue 规范

在提交 issue 时须有对应脚本、提交模式、数据（非必要）、完整日志（重要的东西）等内容

pr 提交规范

1、在 pr 里备注修复的 issue

2、pr commit 模版[hotfix/feat/docs-#issueID][#fix-module] #fix-commit（尽量使用英文，内容描述清楚）

3、修改内容尽量保持与 issue 内容一致，如果出现无关修改，在 pr 中备注出来；

社区文档中心

同时为了帮助社区伙伴更好的了解和使用 ChunJun，我们上线了社区文档中心，欢迎大家使用。

https://dtstack.github.io/chunj