基于大数据开发套件的增量同步策略

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
简介: 因为近期遇到用户在做ETL操作导入数据到MaxCompute的时候,对如何设置数据同步策略有疑惑,所以今天第一波我们来聊一下数据的同步策略,根据数据的特性,看看哪些数据适合增量同步,哪些适合全量同步,又是如何实现的?请认真看完下面的介绍,这些问题都不是事儿。

今天我们来讨论如何使用大数据开发套件进行增量同步。

我们把需要同步的数据,根据数据写入后是否会发生变化,分为会变化的数据(人员表比如说,人员的状态会发生变化)和不会发生变化的数据(一般是日志数据)。针对这两种场景,我们需要设计不同的同步策略。这里以把业务RDS数据库的数据同步到MaxCompute为例做一些说明,其他的数据源的道理是一样的。根据等幂性原则(也就是说一个任务,多次运行的结果是一样的,这样才能支持重跑调度。如果任务出现错误,也比较容易清理脏数据),我每次导入数据都是导入到一张单独的表/分区里,或者覆盖里面的历史记录。

本文的测试时间是2016-11-14,全量同步是在14号做的,同步历史数据到ds=20161113这个分区里。至于本文涉及的增量同步的场景,配置了自动调度,把增量数据在15号凌晨同步到ds=20161114的分区里。数据里有一个时间字段optime,用来表示这条数据的修改时间,从而判断这条数据是否是增量数据。

不变的数据

对应这种场景,因为数据生成后就不会发生变化,我们可以很方便地根据数据的生成规律进行分区,比较常见的是根据日期进行分区,比如每天一个分区。以下是测试数据:

drop table if exists oplog;
create table if not exists oplog(
 optime DATETIME,
 uname varchar(50),
 action varchar(50),
 status varchar(10)
 );

Insert into oplog values(str_to_date('2016-11-11','%Y-%m-%m'),'LiLei','SELECT','SUCCESS');
Insert into oplog values(str_to_date('2016-11-12','%Y-%m-%m'),'HanMM','DESC','SUCCESS');

这里有2条数据,当成历史数据。我先做一次全量数据同步,到昨天的分区里。配置方法如下:
先在MaxCompute创建好表:

--创建好MaxCompute表,按天进行分区
create table if not exists ods_oplog(
 optime datetime,
 uname string,
 action string,
 status string
) partitioned by (ds string);

然后配置了历史数据数据同步:
screenshot

因为只需要跑一次,做以下测试就可以了。测试后到数据开发里把任务的状态改成暂停(最右边的调度配置了)并重新发布,免得明天他继续跑了。之后到MaxCompute里看一下结果:
screenshot
测试通过后。往Mysql里多写一些数据作为增量数据:

 insert into oplog values(CURRENT_DATE,'Jim','Update','SUCCESS');
 insert into oplog values(CURRENT_DATE,'Kate','Delete','Failed'); 
 insert into oplog values(CURRENT_DATE,'Lily','Drop','Failed'); 

然后配置同步任务如下。需要特别注意的是数据过滤这的配置,通过这个配置,可以在15号的凌晨的同步的时候,把14号全天新增的数据查询出来,然后同步到增量分区里。
screenshot
这个任务需要发布,设置调度周期为每天调度,第二天过来一看,MaxCompute里的数据变成了:
screenshot

会变的数据

如人员表、订单表一类的会发生变化的数据,根据数据仓库的4个特点里的反映历史变化的这个特点的要求,我们建议每天对数据进行全量同步。也就是说每天保存的都是数据的全量数据,这样历史的数据和当前的数据都可以很方便地获得。不过如果真实的场景下因为某些特殊情况,需要每天也只做增量同步,因为MaxCompute不支持Update语句来修改数据,只能用别的一些方法来实现。两种同步策略的具体方法如下:

首先我们需要造一些数据:

drop table if exists user ;
create table if not exists user(
    uid int,
    uname varchar(50),
    deptno int,
    gender VARCHAR(1),
    optime DATETIME
    );
--历史数据
insert into user values (1,'LiLei',100,'M',str_to_date('2016-11-13','%Y-%m-%d'));
insert into user values (2,'HanMM',null,'F',str_to_date('2016-11-13','%Y-%m-%d'));
insert into user values (3,'Jim',102,'M',str_to_date('2016-11-12','%Y-%m-%d'));
insert into user values (4,'Kate',103,'F',str_to_date('2016-11-12','%Y-%m-%d'));
insert into user values (5,'Lily',104,'F',str_to_date('2016-11-11','%Y-%m-%d'));
--增量数据
update user set deptno=101,optime=CURRENT_TIME  where uid = 2; --null改成非null
update user set deptno=104,optime=CURRENT_TIME  where uid = 3; --非null改成非null
update user set deptno=null,optime=CURRENT_TIME  where uid = 4; --非null改成null
delete from user where uid = 5;
insert into user(uid,uname,deptno,gender,optime) values (6,'Lucy',105,'F',CURRENT_TIME);

每天全量同步

每天全量同步同步比较简单:

--全量同步
create table ods_user_full(
    uid bigint,
    uname string,
    deptno bigint,
    gender string,
    optime DATETIME 
) partitioned by (ds string);

然后配置同步为:
screenshot
测试后结果为
screenshot
因为每天都是全量同步,没有全量和增量的区别,所以第二天就能看到数据结果为
screenshot
如果需要查询的话,就用where ds =‘20161114’ 来取全量数据即可了。

每天增量

非常不推荐用这种方法,只有在极特殊的场景下才考虑。首先这种场景不支持delete语句,因为被删除的数据无法通过SQL语句的过滤条件查到。当然实际上公司里的代码很少直接有直接删除数据的,都是使用逻辑删除,那delete就转化成update来处理了。但是这里毕竟限制了一些特殊的业务场景不能做了,当出现特殊情况可能导致数据不一致。另外还有一个缺点就是同步后要对新增的数据和历史数据做合并。具体的做法如下:
首先需要创建2张表,一张写当前的最新数据,一张写增量数据:

--结果表
create table dw_user_inc(
    uid bigint,
    uname string,
    deptno bigint,
    gender string,
    optime DATETIME 
);
--增量记录表
create table ods_user_inc(
    uid bigint,
    uname string,
    deptno bigint,
    gender string,
    optime DATETIME 
)

然后全量数据可以直接写入结果表:
screenshot
结果如下:
screenshot
这个只要跑一次的,记得跑好后要暂停掉。
然后把增量数据写入到增量表里:
screenshot
结果如下
screenshot
然后做一次合并

insert overwrite table dw_user_inc 
select 
--所有select操作,如果ODS表有数据,说明发生了变动,以ODS表为准
case when b.uid is not null then b.uid else a.uid end as uid,
case when b.uid is not null then b.uname else a.uname end as uname,
case when b.uid is not null then b.deptno else a.deptno end as deptno,
case when b.uid is not null then b.gender else a.gender end as gender,
case when b.uid is not null then b.optime else a.optime end as optime
from 
dw_user_inc a 
full outer join ods_user_inc b
on a.uid  = b.uid ;

最终结果是:
screenshot
可以看到,delete的那条记录没有同步成功。

对比以上两种同步方式,可以很清楚看到两种同步方法的区别和优劣。第二种方法的优点是同步的增量数据量比较小,但是带来的缺点有可能有数据不一致的风险,而且还需要用额外的计算进行数据合并。如无必要,会变化的数据就使用方法一即可。如果对历史数据希望只保留一定的时间,超出时间的做自动删除,可以设置Lifecycle。

相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps 
目录
相关文章
|
2月前
|
存储 大数据 API
大数据隐私保护策略:加密、脱敏与访问控制实践
【4月更文挑战第9天】本文探讨了大数据隐私保护的三大策略:数据加密、数据脱敏和访问控制。数据加密通过加密技术保护静态和传输中的数据,密钥管理确保密钥安全;数据脱敏通过替换、遮蔽和泛化方法降低敏感信息的敏感度;访问控制则通过用户身份验证和权限设置限制数据访问。示例代码展示了数据库、文件系统和API访问控制的实施方式,强调了在实际应用中需结合业务场景和平台特性定制部署。
507 0
|
10月前
|
存储 大数据
亚信AISWare DataOS大数据中台套件介绍03——dataos开发目录配置
DataOS是一款汇总大部分大数据套件的企业型中台,可以满足企业大部分对大数据存储、计算、稽核的需求。但是平台使用体检并不好,所以有条件还是推荐使用阿里云的大数据组件
333 0
|
1月前
|
数据采集 监控 大数据
大数据时代的数据质量与数据治理策略
在大数据时代,高质量数据对驱动企业决策和创新至关重要。然而,数据量的爆炸式增长带来了数据质量挑战,如准确性、完整性和时效性问题。本文探讨了数据质量的定义、重要性及评估方法,并提出数据治理策略,包括建立治理体系、数据质量管理流程和生命周期管理。通过使用Apache Nifi等工具进行数据质量监控和问题修复,结合元数据管理和数据集成工具,企业可以提升数据质量,释放数据价值。数据治理需要全员参与和持续优化,以应对数据质量挑战并推动企业发展。
206 3
|
29天前
|
分布式计算 大数据 关系型数据库
MaxCompute产品使用问题之如何实现MySQL的实时增量同步
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
29天前
|
分布式计算 DataWorks 大数据
MaxCompute产品使用问题之DataWorks整库全增量同步任务的源库如果新增了表,如何能将这个表快速同步进maxcompute
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
1月前
|
存储 资源调度 大数据
云计算在大数据分析中的弹性资源调度策略
云计算在大数据分析中的弹性资源调度策略
|
2月前
|
数据采集 供应链 安全
利用大数据优化业务流程:策略与实践
【5月更文挑战第11天】本文探讨了利用大数据优化业务流程的策略与实践,包括明确业务目标、构建大数据平台、数据采集整合、分析挖掘及流程优化。通过实例展示了电商和制造企业如何利用大数据改进库存管理和生产流程,提高效率与客户满意度。随着大数据技术进步,其在业务流程优化中的应用将更加广泛和深入,企业需积极采纳以适应市场和客户需求。
|
2月前
|
缓存 运维 NoSQL
面试分享:Redis在大数据环境下的缓存策略与实践
【4月更文挑战第10天】探索Redis在大数据缓存的关键作用,本文分享面试经验及必备知识点。聚焦Redis数据结构(String、List、Set、Hash、Sorted Set)及其适用场景,缓存策略(LRU、LFU、TTL)与过期机制,集群和数据分片,以及性能优化和运维技巧。通过代码示例深入理解,助你面试成功,构建高效缓存服务。
92 4
|
2月前
|
分布式计算 DataWorks 数据库
DataWorks操作报错合集之DataWorks使用数据集成整库全增量同步oceanbase数据到odps的时候,遇到报错,该怎么处理
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
50 0
|
2月前
|
资源调度 分布式计算 大数据
【云计算与大数据技术】资源管理、调度模型策略的讲解
【云计算与大数据技术】资源管理、调度模型策略的讲解
253 0