ChunJun 是⼀款稳定、易⽤、⾼效、批流⼀体的数据集成框架，基于计算引擎 Flink 实现多种异构数据源之间的数据同步与计算。ChunJun 可以把不同来源、格式、特点性质的数据在逻辑上或物理上有机地集中，从⽽为企业提供全⾯的数据共享，目前已在上千家公司部署且稳定运⾏。

在之前，我们曾经为大家介绍过如何利用 ChunJun 实现数据实时同步（点击看正文），本篇将为大家介绍姊妹篇，如何利⽤ ChunJun 实现数据的离线同步。

ChunJun 离线同步案例

离线同步是 ChunJun 的⼀个重要特性，下⾯以最通⽤的 mysql -> hive 的同步任务来介绍离线同步。

配置环境

找⼀个空⽬录，接下来要配置 Flink 和 ChunJun 的环境，下⾯以 /root/chunjun_demo/ 为例⼦。

● 配置 Flink

● 配置 ChunJun

解压好的 ChunJun 有如下⽬录： bin chunjun-dist chunjun-examples lib

● 配置环境变量

● 在 Yarn 上⾯启动 Flink Session

输出如下：

下⾯提交任务会⽤到 Flink Session 这个 Yarn Application Id (application_1683599622970_0270)。

● 其他配置

如果⽤ parquet 格式，需要把 flink-parquet_2.12-1.12.7.jar 放⼊到 flink/lib 下⾯, 在上⾯的例⼦中，需要放到 $FLINK_HOME/lib ⾥⾯。

提交任务

● 在 MySQL 准备数据

如果没有 MySQL 的话，可以⽤ docker 快速创建⼀个。

● 创建 Hive 表

● 在当前⽬录( /root/chunjun_demo/ ) 配置⼀个任务 mysql_hdfs.json

vim mysql_hdfs.json 输⼊如下内容：

因为我们要将 MySQL 同步到 Hive ⾥⾯，但是如果直接同步 Hive 的话，内部会⽤ jdbc，⽽ jdbc 的效率不⾼，因此我们可以直接把数据同步到 Hive 底层的 HDFS 上⾯，所以 writer ⽤到了 hdfswriter。脚本解析如下：

● 提交任务

● 查看任务

任务同步完成, 可以看⼀下 HDFS 上⾯的数据。

查看⼀下 Hive 表的数据。

注意, 如果是分区的 Hive 表，需要⼿动刷新⼀下 Hive 的元数据, 使⽤ MSCK 命令。（MSCK 是 Hive 中的⼀个命令，⽤于检查表中的分区，并将其添加到 Hive 元数据中）

ChunJun 离线同步原理解析

HDFS 文件同步原理

· 对于⽂件系统，同步的时候会先把⽂件写⼊到 path + [filename] ⽬录⾥⾯的 .data 的⽂件⾥⾯，如果任务失败，那么 .data ⾥⾯的⽂件不会⽣效。

· 在 TaskManager 上⾯所有 task 任务结束的时候，会在 JobManager 执⾏ FinalizeOnMaster 的 finalizeGlobal ⽅法, 最终会调⽤到 moveAllTmpDataFileToDir , 把 .data ⾥⾯的⽂件移除到 .data 的上⼀层。

增量同步

增量同步主要针对某些只有 Insert 操作的表，随着业务增⻓，表内数据越来越多。如果每次都同步整表的话，消耗的时间和资源会⽐较多。因此需要⼀个增量同步的功能，每次只读取增加部分的数据。

● 实现原理

其实现原理实际上就是配合增量键在查询的 sql 语句中拼接过滤条件，⽐如 where id > ? ，将之前已经读取过的数据过滤出去。

增量同步是针对于两个及以上的同步作业来说的。对于初次执⾏增量同步的作业⽽⾔，实际上是整表同步，不同于其他作业的在于增量同步作业会在作业执⾏完成后记录⼀个 endLocation 指标，并将这个指标上传到 prometheus 以供后续使⽤。

除第⼀次作业外，后续的所有增量同步作业都会取上⼀次作业的 endLocation 做为本次作业的过滤依据（startLocation）。⽐如第⼀次作业执⾏完后，endLocation 为10，那么下⼀个作业就会构建出例如 SELECT id,name,age from table where id > 10 的 SQL 语句，达到增量读取的⽬的。

● 使用限制

· 只有 RDB 的 Reader 插件可以使⽤

· 通过构建SQL过滤语句实现，因此只能⽤于RDB插件

· 增量同步只关⼼读，不关⼼写，因此只与Reader插件有关

· 增量字段只能为数值类型和时间类型

· 指标需要上传到 prometheus，⽽ prometheus 不⽀持字符串类型，因此只⽀持数据类型和时间类型，时间类型会转换成时间戳后上传

· 增量键的值可以重复，但必须递增

· 由于使⽤ '>' 的缘故，要求字段必须递增

断点续传

断点续传是为了在离线同步的时候，针对⻓时间同步任务如超过1天，如果在同步过程中由于某些原因导致任务失败，从头再来的话成本⾮常⼤，因此需要⼀个断点续传的功能从任务失败的地⽅继续。

● 实现原理

· 基于 Flink 的 checkpoint，在 checkpoint 的时候 会存储 source 端最后⼀条数据的某个字段值，sink 端插件执⾏事务提交。

· 在任务失败，后续通过 checkpoint 重新运⾏时，source 端在⽣成 select 语句的时候将 state ⾥的值作为条件拼接进⾏数据的过滤，达到从上次失败位点进⾏恢复。

· jdbcInputFormat 在拼接读取 SQL 时，如果从 checkpoint 恢复的 state 不为空且 restoreColumn 不为空，则此时会将 checkpoint ⾥的 state 作为起点开始读取数据。

● 适用场景

通过上述原理我们可以知道 source 端必须是 RDB 类型插件，因为是通过 select 语句拼接 where 条件进⾏数据过滤达到断点续传的，同时断点续传需要指定⼀个字段作为过滤条件，且此字段要求是递增的。

· 任务需要开启 checkpoint

· reader 为 RDB 的插件均⽀持且 writer ⽀持事务的插件(如 rdb filesystem 等)，如果下游是幂等性则 writer 插件也不需要⽀持事务

· 作为断点续传的字段在源表⾥的数据是递增的，因为过滤条件是 >

《数据治理行业实践白皮书》下载地址：https://fs80.cn/l134d5 

《数栈V6.0产品白皮书》下载地址：https://fs80.cn/cw0iw1

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=szbzhan

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