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

Posted 数栈DTinsight

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了技术干货|如何利用 ChunJun 实现数据离线同步?相关的知识,希望对你有一定的参考价值。

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

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

ChunJun 离线同步案例

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

配置环境

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

● 配置 Flink

下载 Flink

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

● 配置 ChunJun

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

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

● 配置环境变量

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

● 在 Yarn 上⾯启动 Flink Session

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

输出如下:

echo "stop" | $FLINK_HOME/bin/yarn-session.sh -id application_1683599622970_0270
If this should not be possible, then you can also kill Flink via YARN\'s web interface or via:
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 准备数据

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

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

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

● 创建 Hive 表

CREATE DATABASE IF NOT EXISTS chunjun;
USE chunjun;
-- 创建⼀个名为orders的表,⽤于存储订单信息
CREATE TABLE IF NOT EXISTS chunjun.orders (
 id INT,
 order_id VARCHAR(50),
 user_id INT,
 product_id INT,
 quantity INT,
 order_date TIMESTAMP
)
 STORED AS PARQUET;
-- 查看 hive 表,底层的 HDFS ⽂件位置,下⾯的 SQL 结果⾥⾯ Location 字段,就是 HDFS ⽂件的位置。
desc formatted chunjun.orders;
-- Location: hdfs://ns1/dtInsight/hive/warehouse/chunjun.db/orders
-- ⼀会配置同步任务的时候会⽤到 hdfs://ns1/dtInsight/hive/warehouse/chunjun.db/orders

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

vim mysql_hdfs.json 输⼊如下内容:


"job": 
"content": [
 
"reader": 
"parameter": 
"connection": [
 
"schema": "chunjun",
"jdbcUrl": [ "jdbc:mysql://172.16.85.200:3306/chunjun" ],
"table": [ "orders" ]
 
 ],
"username": "root",
"password": "123456",
"column": [
  "name": "id", "type": "INT" ,
  "name": "order_id", "type": "VARCHAR" ,
  "name": "user_id", "type": "INT" ,
  "name": "product_id", "type": "INT" ,
  "name": "quantity", "type": "INT" ,
  "name": "order_date", "type": "TIMESTAMP" 
 ]
 ,
"name": "mysqlreader"
 ,
"writer": 
"parameter": 
"path": "hdfs://ns1/dtInsight/hive/warehouse/chunjun.db/orders",
"defaultFS": "hdfs://ns1",
"hadoopConfig": 
"dfs.nameservices": "ns1",
"dfs.ha.namenodes.ns1": "nn1,nn2",
"dfs.namenode.rpc-address.ns1.nn1": "172.16.85.194:9000",
"dfs.namenode.rpc-address.ns1.nn2": "172.16.85.200:9000",
"dfs.client.failover.proxy.provider.ns1":
"org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
 ,
"column": [
  "name": "id", "type": "INT" ,
  "name": "order_id", "type": "VARCHAR" ,
  "name": "user_id", "type": "INT" ,
  "name": "product_id", "type": "INT" ,
  "name": "quantity", "type": "INT" ,
  "name": "order_date", "type": "TIMESTAMP" 
 ],
"writeMode": "overwrite",
"encoding": "utf-8",
"fileType": "parquet",
"fullColumnName":
 [ "id", "order_id", "user_id", "product_id", "quantity", "order_date"],
"fullColumnType":
 [ "INT", "VARCHAR", "INT", "INT", "INT", "TIMESTAMP" ]
 ,
"name": "hdfswriter"
 
 
 ],
"setting": 
"errorLimit": 
"record": 0
 ,
"speed": 
"bytes": 0,
"channel": 1
 
 
 

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


"job": 
"content": [
 
"reader": 
"parameter": 
"connectionComment": "数据库链接, 数据库, 表, 账号, 密码",
"connection": [
 
"schema": "chunjun",
"jdbcUrl": [ "jdbc:mysql://172.16.85.200:3306/chunjun" ],
"table": [ "orders" ]
 
 ],
"username": "root",
"password": "123456",
"columnComment": "要同步的列选择, 可以选择部分列",
"column": [
  "name": "id", "type": "INT" ,
  "name": "order_id", "type": "VARCHAR" ,
  "name": "user_id", "type": "INT" ,
  "name": "product_id", "type": "INT" ,
  "name": "quantity", "type": "INT" ,
  "name": "order_date", "type": "TIMESTAMP" 
 ]
 ,
"nameComment" : "source 是 mysql",
"name": "mysqlreader"
 ,
"writer": 
"parameter": 
"pathComment": "HDFS 上⾯的路径, 通过 hive 语句的 desc formatted 查看",
"path": "hdfs://ns1/dtInsight/hive/warehouse/chunjun.db/orders",
"defaultFS": "hdfs://ns1",
"hadoopConfigComment": "是 hdfs ⾼可⽤最基本的配置, 在 Hadoop 配置⽂件 hdfs-site.xml 可以找到",
"hadoopConfig": 
"dfs.nameservices": "ns1",
"dfs.ha.namenodes.ns1": "nn1,nn2",
"dfs.namenode.rpc-address.ns1.nn1": "172.16.85.194:9000",
"dfs.namenode.rpc-address.ns1.nn2": "172.16.85.200:9000",
"dfs.client.failover.proxy.provider.ns1":
"org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
 ,
"columnComment": "要同步的列选择, 可以选择部分列",
"column": [
  "name": "id", "type": "INT" ,
  "name": "order_id", "type": "VARCHAR" ,
  "name": "user_id", "type": "INT" ,
  "name": "product_id", "type": "INT" ,
  "name": "quantity", "type": "INT" ,
  "name": "order_date", "type": "TIMESTAMP" 
 ],
"writeModeComment": "覆盖写⼊到 hdfs 上⾯的⽂件, 可选 overwrite, append(默认模式)",
"writeMode": "overwrite",
"encoding": "utf-8",
"fileTypeComment": "可选 orc, parquet, text",
"fileType": "parquet",
"fullColumnNameComment": "全部字段,有时候 column ⾥⾯同步部分字段,但是⼜需要有全部字段的格式,例如 fileType : text ",
"fullColumnName": [ "id", "order_id", "user_id", "product_id", "quantity", "order_date"], 
"fullColumnTypeComment": "全部字段的类型",
"fullColumnType": [ "INT", "VARCHAR", "INT", "INT", "INT", "TIMESTAMP" ]
 ,
"nameComment" : "sink 是 hdfs",
"name": "hdfswriter"
 
 
 ],
"setting": 
"errorLimit": 
"record": 0
 ,
"speed": 
"bytes": 0,
"channel": 1
 
 
 

● 提交任务

bash chunjun/bin/chunjun-yarn-session.sh -job mysql_hdfs.json -confProp
\\"yarn.application.id\\":\\"application_1683599622970_0270\\"

● 查看任务


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

查看⼀下 Hive 表的数据。

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

MSCK REPAIR TABLE my_table;

ChunJun 离线同步原理解析

HDFS 文件同步原理

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

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

public interface FinalizeOnMaster

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

// 在 JobManager 执⾏
@Override
protected void moveAllTmpDataFileToDir() 
if (fs == null) 
openSource();
 
String currentFilePath = "";
try 
Path dir = new Path(outputFilePath);
Path tmpDir = new Path(tmpPath);

FileStatus[] dataFiles = fs.listStatus(tmpDir);
for (FileStatus dataFile : dataFiles) 
currentFilePath = dataFile.getPath().getName();
fs.rename(dataFile.getPath(), dir);
LOG.info("move temp file: to dir:", dataFile.getPath(), dir);
 
fs.delete(tmpDir, true);
  catch (IOException e) 
throw new ChunJunRuntimeException(
String.format(
"can\'t move file:[%s] to dir:[%s]", currentFilePath, outputFilePath),
e);
 

增量同步

增量同步主要针对某些只有 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

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

【作者】刘俊源,任职于广州九一乐维。


场景需求

大部分企业都会存在测试数据库。然而,这些公司的运维人员普遍的做法是编写一个shell脚本通过mysqldump,这个MySQL数据库备份工具结合Linux的crontab这个定时任务命令,实现每周或者每月对数据库进行备份,如果不监控这些测试数据库每周是否备份成功,很可能会在后期的工作中出现问题。因此,今天围绕这个话题,介绍一下如何通过Zabbix监控mysqldump数据库的备份。

实验过程

1. 在mysql数据库创建一个演示用的,名为test的数据库,在test数据库创建一个test表。

Mysql的语句

Create database test; #创建名为test的数据库

Use test; #使用test数据库

create table test (name varchar(50)); #创建一个只有name字段的test表格

1.1 创建测试数据库test

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

1.2 编写数据库备份脚本

然后是编写一个简单备份数据库的shell脚本(bak.sh),在crontab定时任务中,设置脚本执行的时间

正常来说(备份的数据库文件名,必须时数据库名加上数据库备份时间命名)

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

Mysqldump –u用户名 –p密码 –h数据库ip 数据库名(test) > 备份的数据库存放的位置

如果数据库备份成功,在/tmp/bak_result.txt就会是一个0。反之,非0就是数据库备份失败。

注意:存放数据库备份的位置最好不要是用户目录,否则其他用户可能无法查看,因此选择/tmp目录用来测试

最后要给bak.sh加上执行权限,否则定时任务无法执行

Chmod o+x bak.sh

1.3 设置定时计划,为了方便演示效果(备份时间设置为1分钟备份一次)

Crontab –e打开定时任务

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

查看是否备份成功

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶
2.创建监控项

在zabbix_agent.conf配置自定义键值

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

重启zabbix_agent

在zabbix_server配置监控项

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶
如上图所示,mysql备份的值为0,也就是备份成功。

创建告警。如果值非0,那么就是数据库备份失败

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

为了验证备份失败。我们可以执行删除数据库test的操作。那么mysql是否备份成功这个监控项的值就会不为0,接着会出现告警。

删除数据库命令(drop database 数据库名)

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

此时,mysql是否备份成功的监控项值不为0。导出的数据库文件也是无效的。

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

触发器也出现告警。

实验完毕。

原题:[技术干货]利用 zabbix 监控 mysqldump 定时备份数据库是否成功
如有任何问题,可点击文末 阅读原文 ,到社区原文下评论交流
觉得本文有用,请转发或点击“在看”,让更多同行看到


 资料/文章推荐:



  • Zabbix教程:从入门到精通

    http://www.talkwithtrend.com/Document/detail/tid/179485


http://www.talkwithtrend.com/Topic/3937


下载 twt 社区客户端 APP

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

或到应用商店搜索“twt”


以上是关于技术干货|如何利用 ChunJun 实现数据离线同步?的主要内容,如果未能解决你的问题,请参考以下文章

DTMO直播预告丨ChunJun 2022年开源规划&支持异构数据源

技术干货:利用 Zabbix 监控 mysqldump 定时备份数据库是否成功 | 运维进阶

纯钧chunjun的http-x插件修复

保姆级教程!玩转 ChunJun 详细指南

干货 | 快速实现数据导入及简单DCS的实现

袋鼠云批流一体分布式同步引擎ChunJun(原FlinkX)的前世今生