背景:一些客户反馈,增量同步数据到MaxCompute按照全天的数据做增量数据同步数据量太大,且不使用按天的增量同步数据,进行在MaxCompute上进行数据处理得出增量数据对于delete的相关数据不能做到很好的支持,在次给大家一个对增量数据同步的方案使用DTS做增量同步数据到MaxCompute,数据源为ECS上自建的mysql5.7。
一、为自建MySQL创建账号并设置
1.1登陆自建Mysql数据库
1.2创建mysql数据库中用于数据迁移/同步的账号
CREATE USER 'dtsmigration'@'%' IDENTIFIED BY 'Dts123456';
说明:
- username:待创建的账号。
- host:允许该账号登录的主机,如果允许该账号从任意主机登录数据库,可以使用百分号(%)。
- password:账号的密码。
1.3对账号进行授权操作
GRANT privileges ON databasename.tablename TO 'username'@'host' WITH GRANT OPTION;
说明:
- privileges:授予该账号的操作权限,如SELECT、INSERT、UPDATE等,如果要授予该账号所有权限,则使用ALL。
- databasename:数据库名。如果要授予该账号具备所有数据库的操作权限,则使用星号(*)。
- tablename:表名。如果要授予该账号具备所有表的操作权限,则使用星号(*)。
- username:待授权的账号。
- host:允许该账号登录的主机,如果允许该账号从任意主机登录,则使用百分号(%)。
- WITH GRANT OPTION:授予该账号使用GRANT命令的权限,该参数为可选。
如果要给账户赋予所有数据库和表的权限,并容许从任意主机登陆数据库
GRANT ALL ON *.* TO 'dtsmigration'@'%';
1.4开启并设置自建Mysql数据库binlog
a.使用vim命令,修改配置文件my.cnf中的如下参数
log_bin=mysql_bin
binlog_format=row
server_id=2 //设置大于1的整数
binlog_row_image=full //当自建MySQL的版本大于5.6时,则必须设置该项。
b.修改完成后,重启Mysql进程。
service mysqld restart
二、同步过程介绍
2.1结构初始化
DTS将源库中待同步表的结构定义信息同步至MaxCompute中,初始化时DTS会为表名增加_base后缀。例如源表为customer,那么MaxCompute中的表即为customer_base。
2.2全量数据初始化
DTS将源库中待同步表的存量数据,全部同步至MaxCompute中的目标表名_base表中(例如从源库的customer表同步至MaxCompute的customer_base表),作为后续增量同步数据的基线数据。
2.3增量数据同步
DTS在MaxCompute中创建一个增量日志表,表名为同步的目标表名_log,例如customer_log,然后将源库产生的增量数据实时同步到该表中。
三、增量同步实践
3.1购买DTS同步
3.2查看购买的DTS同步,点击配置同步链路
3.3配置对应的数据源和相应的MaxCompute项目
3.4点击授予权限的同步账号操作
3.5选择对应的增量同步数据的同步实践,并选择需要同步的表
3.6同步配置预检查
3.7查询同步的全量数据
3.8查看同步成功的增量数据分区user_log
3.9查看增量数据同步的数据
元数据的字段介绍
字段 | 说明 |
---|---|
record_id | 增量日志的记录id,为该日志唯一标识。 说明 - id的值唯一且递增。 - 如果增量日志的操作类型为UPDATE,那么增量更新会被拆分成两条记录,且record_id的值相同。 |
operation_flag | 操作类型,取值: - I:INSERT操作。 - D:DELETE操作。 - U:UPDATE操作。 |
utc_timestamp | 操作时间戳,即binlog的时间戳(UTC 时间)。 |
before_flag | 所有列的值是否为更新前的值,取值:Y或N。 |
after_flag | 所有列的值是否为更新后的值,取值:Y或N。 |
四、根据时间点位,整合该时间点位之前的全量数据
4.1建立全量数据表
CREATE TABLE IF NOT EXISTS maxcomputeone_dev.user_all(uid BIGINT,uname STRING,deptno BIGINT,gender STRING,optime DATETIME,record_id BIGINT,operation_flag STRING,utc_timestamp BIGINT,before_flag STRING,after_flag STRING);
4.2查看增量数据最后同步的点位,最后整合全量数据到user_all
合并语句
set odps.sql.allow.fullscan=true;
insert overwrite table user_all
select uid,
uname,
deptno,
gender,
optime
from(
select row_number() over(partition by t.uid
order by record_id desc, after_flag desc) as record_num, record_id, operation_flag, after_flag, uid, uname, deptno,gender,optime
from(
select incr.record_id, incr.operation_flag, incr.after_flag, incr.uid, incr.uname,incr.deptno,incr.gender,incr.optime
from user_log incr
where utc_timestamp <= 1585107804
union all
select 0 as record_id, 'I' as operation_flag, 'Y' as after_flag, base.uid, base.uname,base.deptno,base.gender,base.optime
from user_base base) t) gt
where record_num=1
and after_flag='Y';
欢迎加入“MaxCompute开发者社区2群”,点击链接申请加入或扫描二维码
https://h5.dingtalk.com/invite-page/index.html?bizSource=____source____&corpId=dingb682fb31ec15e09f35c2f4657eb6378f&inviterUid=E3F28CD2308408A8&encodeDeptId=0054DC2B53AFE745