技术干货｜如何利用 ChunJun 实现数据离线同步？

ChunJun 是⼀款稳定、易⽤、⾼效、批流⼀体的数据集成框架，基于计算引擎 Flink 实现多种异构数据源之间的数据同步与计算。ChunJun 可以把不同来源、格式、特点性质的数据在逻辑上或物理上有机地集中，从⽽为企业提供全⾯的数据共享，目前已在上千家公司部署且稳定运⾏。
在之前，我们曾经为大家介绍过如何利用 ChunJun 实现数据实时同步（点击看正文），本篇将为大家介绍姊妹篇，如何利⽤ ChunJun 实现数据的离线同步。
ChunJun 离线同步案例

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

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

1. wget "http://archive.apache.org/dist/flink/flink-1.12.7/flink-1.12.7-bin-scala_2.12.tgz"
2. tar -zxvf chunjun-dist.tar.gz

复制代码

● 配置 ChunJun

1. #下载 chunjun, 内部依赖 flink 1.12.7
2. wget https://github.com/DTStack/chunjun/releases/download/v1.12.8/chunjun-dist-1.12-SNAPSHOT.tar.gz
3. #新创建⼀个⽬录
4. mkdir chunjun && cd chunjun
5. #解压到指定⽬录
6. tar -zxvf chunjun-dist-1.12-SNAPSHOT.tar.gz

复制代码

解压好的 ChunJun 有如下⽬录：
bin
chunjun-dist
chunjun-examples
lib
● 配置环境变量

1. #配置 Flink 环境变量
2. echo "FLINK_HOME=/root/chunjun_demo/flink-1.12.7" >> /etc/profile.d/sh.local
3. #配置 Chunjun 的环境变量
4. echo "CHUNJUN_DIST=/root/chunjun_demo/chunjun/chunjun-dist" >> /etc/profile.d/sh.local
5. #刷新换新变量
6. . /etc/profile.d/sh.local

复制代码

● 在 Yarn 上⾯启动 Flink Session

1. #启动 Flink Session
2. bash $FLINK_HOME/bin/yarn-session.sh -t $CHUNJUN_DIST -d

复制代码

输出如下：

1. echo "stop" | $FLINK_HOME/bin/yarn-session.sh -id application_1683599622970_0270
2. If this should not be possible, then you can also kill Flink via YARN's web interface or via:
3. yarn application -kill application_1683599622970_0270

复制代码

下⾯提交任务会⽤到 Flink Session 这个 Yarn Application Id (application_1683599622970_0270)。
● 其他配置
如果⽤ parquet 格式，需要把 flink-parquet_2.12-1.12.7.jar 放⼊到 flink/lib 下⾯, 在上⾯的例⼦中，需要放到 $FLINK_HOME/lib ⾥⾯。

提交任务

● 在 MySQL 准备数据

1. -- 创建⼀个名为ecommerce_db的数据库，⽤于存储电商⽹站的数据
2. CREATE DATABASE IF NOT EXISTS chunjun;
3. USE chunjun;
4. -- 创建⼀个名为orders的表，⽤于存储订单信息
5. CREATE TABLE IF NOT EXISTS orders (
6. id INT AUTO_INCREMENT PRIMARY KEY, -- ⾃增主键
7. order_id VARCHAR(50) NOT NULL, -- 订单编号，不能为空
8. user_id INT NOT NULL, -- ⽤户ID，不能为空
9. product_id INT NOT NULL, -- 产品ID，不能为空
10. quantity INT NOT NULL, -- 订购数量，不能为空
11. order_date TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
12. -- 订单⽇期，默认值为当前时间戳，不能为空
13. );
14. -- 插⼊⼀些测试数据到orders表
15. INSERT INTO orders (order_id, user_id, product_id, quantity)
16. VALUES ('ORD123', 1, 101, 2),
17.        ('ORD124', 2, 102, 1),
18.        ('ORD125', 3, 103, 3),  
19.        ('ORD126', 1, 104, 1),
20.        ('ORD127', 2, 105, 5);
21.       
22. select * from chunjun.orders;      

复制代码

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

1. docker pull mysql:8.0.12
2. docker run --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=123456 -d mysql:8.0.12

复制代码

● 创建 Hive 表

1. CREATE DATABASE IF NOT EXISTS chunjun;
2. USE chunjun;
3. -- 创建⼀个名为orders的表，⽤于存储订单信息
4. CREATE TABLE IF NOT EXISTS chunjun.orders (
5. id INT,
6. order_id VARCHAR(50),
7. user_id INT,
8. product_id INT,
9. quantity INT,
10. order_date TIMESTAMP
11. )
12. STORED AS PARQUET;

复制代码

1. -- 查看 hive 表，底层的 HDFS ⽂件位置，下⾯的 SQL 结果⾥⾯ Location 字段，就是 HDFS ⽂件的位置。
2. desc formatted chunjun.orders;
3. -- Location: hdfs://ns1/dtInsight/hive/warehouse/chunjun.db/orders
4. -- ⼀会配置同步任务的时候会⽤到 hdfs://ns1/dtInsight/hive/warehouse/chunjun.db/orders

复制代码

● 在当前⽬录( /root/chunjun_demo/ ) 配置⼀个任务 mysql_hdfs.json
vim mysql_hdfs.json 输⼊如下内容：

1. {
2. "job": {
3. "content": [
4. {
5. "reader": {
6. "parameter": {
7. "connection": [
8. {
9. "schema": "chunjun",
10. "jdbcUrl": [ "jdbc:mysql://172.16.85.200:3306/chunjun" ],
11. "table": [ "orders" ]
12. }
13. ],
14. "username": "root",
15. "password": "123456",
16. "column": [
17. { "name": "id", "type": "INT" },
18. { "name": "order_id", "type": "VARCHAR" },
19. { "name": "user_id", "type": "INT" },
20. { "name": "product_id", "type": "INT" },
21. { "name": "quantity", "type": "INT" },
22. { "name": "order_date", "type": "TIMESTAMP" }
23. ]
24. },
25. "name": "mysqlreader"
26. },
27. "writer": {
28. "parameter": {
29. "path": "hdfs://ns1/dtInsight/hive/warehouse/chunjun.db/orders",
30. "defaultFS": "hdfs://ns1",
31. "hadoopConfig": {
32. "dfs.nameservices": "ns1",
33. "dfs.ha.namenodes.ns1": "nn1,nn2",
34. "dfs.namenode.rpc-address.ns1.nn1": "172.16.85.194:9000",
35. "dfs.namenode.rpc-address.ns1.nn2": "172.16.85.200:9000",
36. "dfs.client.failover.proxy.provider.ns1":
37. "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
38. },
39. "column": [
40. { "name": "id", "type": "INT" },
41. { "name": "order_id", "type": "VARCHAR" },
42. { "name": "user_id", "type": "INT" },
43. { "name": "product_id", "type": "INT" },
44. { "name": "quantity", "type": "INT" },
45. { "name": "order_date", "type": "TIMESTAMP" }
46. ],
47. "writeMode": "overwrite",
48. "encoding": "utf-8",
49. "fileType": "parquet",
50. "fullColumnName":
51. [ "id", "order_id", "user_id", "product_id", "quantity", "order_date"],
52. "fullColumnType":
53. [ "INT", "VARCHAR", "INT", "INT", "INT", "TIMESTAMP" ]
54. },
55. "name": "hdfswriter"
56. }
57. }
58. ],
59. "setting": {
60. "errorLimit": {
61. "record": 0
62. },
63. "speed": {
64. "bytes": 0,
65. "channel": 1
66. }
67. }
68. }
69. }

复制代码

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

1. {
2. "job": {
3. "content": [
4. {
5. "reader": {
6. "parameter": {
7. "connectionComment": "数据库链接, 数据库, 表, 账号, 密码",
8. "connection": [
9. {
10. "schema": "chunjun",
11. "jdbcUrl": [ "jdbc:mysql://172.16.85.200:3306/chunjun" ],
12. "table": [ "orders" ]
13. }
14. ],
15. "username": "root",
16. "password": "123456",
17. "columnComment": "要同步的列选择, 可以选择部分列",
18. "column": [
19. { "name": "id", "type": "INT" },
20. { "name": "order_id", "type": "VARCHAR" },
21. { "name": "user_id", "type": "INT" },
22. { "name": "product_id", "type": "INT" },
23. { "name": "quantity", "type": "INT" },
24. { "name": "order_date", "type": "TIMESTAMP" }
25. ]
26. },
27. "nameComment" : "source 是 mysql",
28. "name": "mysqlreader"
29. },
30. "writer": {
31. "parameter": {
32. "pathComment": "HDFS 上⾯的路径, 通过 hive 语句的 desc formatted 查看",
33. "path": "hdfs://ns1/dtInsight/hive/warehouse/chunjun.db/orders",
34. "defaultFS": "hdfs://ns1",
35. "hadoopConfigComment": "是 hdfs ⾼可⽤最基本的配置, 在 Hadoop 配置⽂件 hdfs-site.xml 可以找到",
36. "hadoopConfig": {
37. "dfs.nameservices": "ns1",
38. "dfs.ha.namenodes.ns1": "nn1,nn2",
39. "dfs.namenode.rpc-address.ns1.nn1": "172.16.85.194:9000",
40. "dfs.namenode.rpc-address.ns1.nn2": "172.16.85.200:9000",
41. "dfs.client.failover.proxy.provider.ns1":
42. "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
43. },
44. "columnComment": "要同步的列选择, 可以选择部分列",
45. "column": [
46. { "name": "id", "type": "INT" },
47. { "name": "order_id", "type": "VARCHAR" },
48. { "name": "user_id", "type": "INT" },
49. { "name": "product_id", "type": "INT" },
50. { "name": "quantity", "type": "INT" },
51. { "name": "order_date", "type": "TIMESTAMP" }
52. ],
53. "writeModeComment": "覆盖写⼊到 hdfs 上⾯的⽂件, 可选 overwrite, append(默认模式)",
54. "writeMode": "overwrite",
55. "encoding": "utf-8",
56. "fileTypeComment": "可选 orc, parquet, text",
57. "fileType": "parquet",
58. "fullColumnNameComment": "全部字段，有时候 column ⾥⾯同步部分字段，但是⼜需要有全部字段的格式，例如 fileType : text ",
59. "fullColumnName": [ "id", "order_id", "user_id", "product_id", "quantity", "order_date"],
60. "fullColumnTypeComment": "全部字段的类型",
61. "fullColumnType": [ "INT", "VARCHAR", "INT", "INT", "INT", "TIMESTAMP" ]
62. },
63. "nameComment" : "sink 是 hdfs",
64. "name": "hdfswriter"
65. }
66. }
67. ],
68. "setting": {
69. "errorLimit": {
70. "record": 0
71. },
72. "speed": {
73. "bytes": 0,
74. "channel": 1
75. }
76. }
77. }
78. }

复制代码

● 提交任务

1. bash chunjun/bin/chunjun-yarn-session.sh -job mysql_hdfs.json -confProp
2. {"yarn.application.id":"application_1683599622970_0270"}

复制代码

● 查看任务


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

查看⼀下 Hive 表的数据。

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

1. MSCK REPAIR TABLE my_table;

复制代码

ChunJun 离线同步原理解析

HDFS 文件同步原理

· 对于⽂件系统，同步的时候会先把⽂件写⼊到 path + [filename] ⽬录⾥⾯的 .data 的⽂件⾥⾯，如果任务失败，那么 .data ⾥⾯的⽂件不会⽣效。
· 在 TaskManager 上⾯所有 task 任务结束的时候，会在 JobManager 执⾏ FinalizeOnMaster 的 finalizeGlobal ⽅法, 最终会调⽤到 moveAllTmpDataFileToDir , 把 .data ⾥⾯的⽂件移除到 .data 的上⼀层。
public interface FinalizeOnMaster {

1. /**
2. The method is invoked on the master (JobManager) after all (parallel) instances of an OutputFormat finished.
3. Params:parallelism – The parallelism with which the format or functions was run.
4. Throws:IOException – The finalization may throw exceptions, which may cause the job to abort.
5. */
6. void finalizeGlobal(int parallelism) throws IOException;
7. }

复制代码

1. // 在 JobManager 执⾏
2. @Override
3. protected void moveAllTmpDataFileToDir() {
4. if (fs == null) {
5. openSource();
6. }
7. String currentFilePath = "";
8. try {
9. Path dir = new Path(outputFilePath);
10. Path tmpDir = new Path(tmpPath);
12. FileStatus[] dataFiles = fs.listStatus(tmpDir);
13. for (FileStatus dataFile : dataFiles) {
14. currentFilePath = dataFile.getPath().getName();
15. fs.rename(dataFile.getPath(), dir);
16. LOG.info("move temp file:{} to dir:{}", dataFile.getPath(), dir);
17. }
18. fs.delete(tmpDir, true);
19. } catch (IOException e) {
20. throw new ChunJunRuntimeException(
21. String.format(
22. "can't move file:[%s] to dir:[%s]", currentFilePath, outputFilePath),
23. e);
24. }
25. }

复制代码

增量同步

增量同步主要针对某些只有 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://www.dtstack.com/resources/1004?src=szsm
《数据治理行业实践白皮书》下载地址：https://www.dtstack.com/resources/1001?src=szsm
想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=szbky
同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术qun」，交流最新开源技术信息，qun号码：30537511，项目地址：https://github.com/DTStack

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！