我们以MySQL数据增量同步到MaxCompute(ODPS)为例。
首先示例中我们创建了一个MySQL的数据库以及示例表demo_wpw_addsync。
该表有3列,分别存储id、name和时间戳
表内现在有一条8月8日的数据,我们需要配置增量同步规则,将这条数据同步到Max Compute中。
接下来我们进入到DataWorks数据开发界面,使用Di节点配置数据集成同步任务。
可以看到“数据来源”我们配置了刚才的MySQL数据库,“数据去向”是我们的一个ODPS实例里一张名字相同的表,列字段和MySQL保持一致。这里重点是数据过滤的条件配置,这里我们配置了col3 >=’${bizdate} 00:00:00’ 。意思就是源端数据抽取时col3这列的值要大于等于我们配置的值,col3我们存的是数据的时间戳,bizdate是我们配置的一个参数变量。
点开调度配置我们可以看到参数这栏里我们配置bizdate = ${yyyy-mm-dd+1},也就是取“业务时间+1”天,一般离线同步今天同步昨天的数据,示例中当天是8月8日,所以业务时间为8月7日,那么加一天就是8月8日了。
我们将任务提交后,可以去运维中心开发环境测试运行
右键点击节点后点测试
弹出来冒烟测试框,可以看到业务日期是8月7日,我们直接点确定新建实例运行测试
我们点击最新的实例,点击“查看日志”,可以查看执行日志
可以看到实际执行时变量bizdate解析成了8月8日
这样我们实际读取MySQL时,where条件就会解析成col3 >= ’2021-08-08 00:00:00’,也就是创建时时间戳大于8月8日0点的数据都会被读取
我们在DataWorks中可以查询ODPS表,可以看到MySQL中的这条数据已经同步过来了。当然,我们可以设置更加复杂的数据过滤条件,来满足实际使用的场景需要
比如上图中,我们在右侧“参数”配置中,定义了 bizdate、canshu2、bizdate2这三个参数,分别赋值‘${yyyy-mm-dd+1}’、‘$[hh24:mi:ss]’、‘${yyyy-mm-dd+2}’。
最终bizdate=20210808,canshu2= 00:29:00,bizdate2=20210809。这里canshu2取值是等于定时调度时间。
对调度参数不熟悉的用户,调度参数配置规则可以查看这篇文章:
https://developer.aliyun.com/article/784963
那么左侧的数据过滤条件就会解析成col3 >= ‘20210808 00:29:00’ and col3 <= ‘20210809 23:59:59’
也就是8月8日定时调度时间之后到8月9日最后一秒钟(8月10日前)的所有数据。
这里的条件依用户的使用场景定义即可。
增量同步的核心就是数据源存在一个DateTime类型的列,通过定义调度参数以及该列的过滤条件,来实现增量同步。实际任务执行时是每天一次,业务日期自动顺延,那bizdate解析出来的日期也会自动增加。
另外,实时同步的话大体都是增量同步的,需要注意的是配置一个重置位点
通常大家数据上云时会做一次全量同步后增量同步的配置,增量同步任务如果需要修改配置的话可以先暂停任务,记下任务暂停时间,然后重启任务时配置这个“重置位点”,取任务暂停时间,这样,实时同步任务会从重置位点向后同步数据,避免重复消费之前的数据。
DataWorks使用过程中有任何问题,欢迎加入我们的大群咨询:
https://developer.aliyun.com/article/740906