Hadoop 是 Apache 基金会旗下最知名的基础架构开源项目之一。自 2006 年诞生以来，逐步发展成为海量数据存储、处理最为重要的基础组件，形成了非常丰富的技术生态。

作为国内顶尖的 Hadoop 开源生态技术峰会，第四届 China Apache Hadoop Meetup 于 2022 年 9 月 24 日在上海成功举办。

围绕“云数智聚 砥柱笃行”的主题，来自华为、阿里、网易、字节跳动、bilibili、平安银行、袋鼠云、英特尔、Kyligence、Ampere 等多所企业单位，以及来自 Spark、Fluid、ChunJun、Kyuubi、Ozone、IoTDB、Linkis、Kylin、Uniffle 等开源社区的多位嘉宾均参与了分享讨论。

作为此次 Meetup 参与社区之一，也是大数据领域的项目，ChunJun 也带来了一些新的声音：

ChunJun 框架在实时数据采集和还原上的实现和原理是怎样的？这段时间以来，ChunJun 有哪些新发展，对于未来发展又有着怎样的新想法？

作为袋鼠云资深大数据引擎开发专家，徐超带来了他的分享，将从一个独特的角度来介绍 ChunJun 数据集成在数据还原上的探索和实践。

一、ChunJun 框架介绍

第一个问题：ChunJun 这个框架是什么？能干啥？

ChunJun(原 FlinkX) 是袋鼠云基于 Flink 基座自研的数据集成框架,经过 4 年多的迭代，已经成为一个稳定，高效，易用的批流一体的数据集成工具，可实现多种异构数据源高效的数据同步，目前已有 3.2K+Star。

开源项目地址：

https://github.com/DTStack/chunjun

https://gitee.com/dtstack_dev_0/chunjun

01 ChunJun 框架结构

ChunJun 框架基于 Flink 进行开发，提供了丰富的插件，同时添加了断点续传、脏数据管理、数据还原等特性。

02 ChunJun 批量同步

• 支持增量同步

• 支持断点续传

• 支持多通道 &并发

• 支持脏数据（记录和控制）

• 支持限流

• 支持 transformer

03 ChunJun 离线

二、实时数据采集上的实现和原理

01 一个样例

02 ChunJun 插件装载逻辑

03 ChunJun 插件定义

04 ChunJun 数据流转

05 ChunJun 动态执行

面对监听多个表的情况，包括新添加表的数据，我们如何执行下游的写入：

• 支持 Update 转换 before,after

• 添加扩展参数，DB,Schema,Table, ColumnInfo

• 支持动态构建 PreparedStatement

06 ChunJun 间隔轮询

什么是间隔轮询？我们是如何做的？

• 校验轮询字段类型，如果不是数值类型且 source 并行度大于 1，报错不支持

• 创建三个数据分片，startlocation 为 null 或者配置的值，mod 分别为 0，1，2

• 构造 SQL：不同 SQL 的取余函数不同，各自插件实现

select id,name,age from table where (id > ? and ) mod(id, 3) = 0 order by id;

select id,name,age from table where (id > ? and ) mod(id, 3) = 1 order by id;

select id,name,age from table where (id > ? and ) mod(id, 3) = 2 order by id;

• 执行 SQL，查询并更新 lastRow

• 第一次 result 查询完后，若脚本中没有配置 startlocation，则之前的查询 SQL 为：

select id,name,age from table where mod(id, 3) = 1 order by id;

将其更新为：

select id,name,age from table where id > ? and mod(id, 3) = 1 order by id;

• CP 时获取 lastRow 中的 id 值，保存到 state 中

三、实时数据还原上的实现和原理

01 数据还原介绍

数据还原基于对应的数据库的 CDC 采集功能,比如上面提到的 Oracle Logminer,MySQL binglog，支持将捕获到的数据完整的还原到下游，所以不仅仅包括 DML，而且也需要对 DDL 进行监听，将上游数据源的所有变更行为发送到下游数据库的还原。

难点

· DDL，DML 如何有序的发送到下游

· DDL 语句如何根据下游数据源的特性进行对应的操作（异构数据源间 DML 的转换）

· DML 语句中的 insert update, delete 如何进行处理

02 一个样例

03 整体流程

数据从上游的数据源获取之后经过一些列的算子的处理之后按数据在原始表中的顺序准确的还原到目标数据源，完成数据的实时获取链路。

04 DDL 解析

数据还原- DDL 转换

· 基于 Calcite 解析数据源 DdlSql 转为 SqlNode

· SqlNode 转为中间数据 DdlData

· ddlData 转为 sql：不同语法之间互相转换；不同数据源字段类型互相转换

05 名字映射

在实时还原中，当前上下游表字段对应关系必须是相同的，即上游的 database schema table 对应的表只能写入下游 database schema table 相同的表，同时字段名称也必须是相同的。本次迭代将针对表路径可以进行一个自定义映射以及字段类型进行自定义映射。

• db or schema 转换

• 表名称转换

• 字段名(提供大小写转换)，类型隐式转换

06 中间数据缓存

数据（不论 ddl 还是 dml 数据）下发到对应表名下的 unblock 队列中，worker 在轮询过程中，处理 unblock 数据队列中的数据，在遇到 ddl 数据之后，将数据队列置为 block 状态，并将队列引用交给 store 处理。

store 在拿到队列引用之后，将队列头部的 ddl 数据下发到外部存储中，并监听外部存储对 ddl 的反馈情况（监听工作由 store 中额外的线程来执行），此时，队列仍然处于 block 状态。

在收到外部存储的反馈之后，将数据队列头部的 ddl 数据移除，同时将队列状态回归为 unblock 状态，队列引用还给 worker。

07 目标端接收数据

• 获取到 DdlOperator 对象

• 根据目标数据源对应的 DDLConvertImpl 解析器转换为目标数据源 sql

• 执行对应的 sql,比如删除表

• 触发调整 DDLChange 表，修改对应的 DDL 状态

• 中间存储 Restore 算子，监听状态变更，执行后续数据下发操作

四、ChunJun 未来规划

• 提供对 Session 进行管理

• 提供 restful 服务，ChunJun 本身作为一个服务，便于外围系统进行集成

• 对实时数据还原进行加强，包括扩展支持更多的数据源的 DDL 解析

此外，本次分享的全文视频内容也可以随时观看，如果您有兴趣，欢迎前往袋鼠云 B 站平台观看。

Apache Hadoop Meetup 2022

ChunJun 视频回顾：

https://www.bilibili.com/video/BV1sN4y1P7qk/?spm_id_from=333.337.search-card.all.click

袋鼠云开源框架钉钉技术交流群（30537511），欢迎对大数据开源项目有兴趣的同学加入交流最新技术信息，开源项目库地址：https://github.com/DTStack/Taier