如何正确的做增量加工程序员

简介：回到十多年前，增量加工这个方法并不是一种需要特别需要提出的方法，因为关系数据库的存储与计算性能十分有限（即便是MPP数据库平台也不是全都是做全量加工），增量加工是最普遍的方式。本文讲述了如何在MaxCompute上用与关系数据库的不同的方式做增量数据的加工。
1.增量加工
回到十多年前，增量加工这个方法并不是一种需要特别需要提出的方法，因为关系数据库的存储与计算性能十分有限（即便是MPP数据库平台也不是全都是做全量加工），增量加工是最普遍的方式。
数据库系统是支持事务的，ACID（原子性、一致性、隔离性、持久性）四大特性可以完美的支持在一个数据表上同时做更新、删除、插入操作。数据库系统的数据存储是到每一个4K或者8K这种大小的数据块上的，详细的统计信息与索引结构都允许我们高效来做增量数据处理。
1.1.问题简述
在当前的MaxCompute这种分布式文件系统上，这些操作都变得不容易了。我们的数据块已经是64MB，不是KB这个量级。我们也没有索引这种加速从一千万数据中找到五十行数据的结构。
那么我们怎么在MaxCompute做增量加工呢？说实话，不太好做。因为没有索引结构，我们每一次的处理都是全量数据检索。如果还是跟之前在关系数据库一样频繁的提交，不但无法体现增量加工的性能与资源优势，反而成为了劣势。（如果我们还想使用关系数据库支持的delete、update这些特性，可以看下MaxCompute公共云近期上线的新特性“Transactional表”。）
那么我们要不要做呢？总结一句话：能做的地方还是可以做一下，但是不要勉强，不要大规模的去做，毕竟做增量加工不容易。
2.解决方案
增量加工的前提是我们获取到了增量数据，相比全量数据增量数据是一个更小的集合，然后我们希望利用这个小增量集合来完成数据加工的过程而不是使用全量，这样就可以更快速、更节约的完成整个数据加工过程。
但是增量加工在MaxCompute总结为两个场景：
场景一，全量加工所需资源无法满足时效性要求，性能急需优化；
场景二，增量加工逻辑简单，相比全量加工性能优势明显；
2.1.加工原则
然后我们需要确立一些使用增量加工的原则，突破或者不遵守这些原则都是不合理或者不正确的。
一、增量表（增量状态[U\D\I\K]，数据更新时间）；
二、2张增量表不能直接关联，必须要有至少一张表是全量；
三、增量加工产出的结果表，还需要记录增量状态和数据更新时间；
四、多个表关联情况下，需要取多个表的增量标识，只要某一个表的关联行是增量就使用该表增量标识；
五、只有主表或则INNER JOIN的表的INSERT和DELETE状态可以传递到下一层，其他表的增量状态都是UPDATE；
2.2.MERGE逻辑
增量集成到MaxCompute平台的数据落地后，需要做一次MERGE才会产生ODS层的全量数据。所以，MERGE逻辑是最简单和经典的增量加工逻辑。最简单的MERGE逻辑如下：
INSERT OVERWRITE TABLE table_old PARTITION(ds='${ds}')
SELECT (ds)?+.+
FROM table_old a --全量表
LEFT ANTI JOIN
table_new b --增量表
ON a.pk = b.pk
AND b.ds = '${ds}'
WHERE a.ds = '${lastdate}'
UNION ALL
SELECT b.*
FROM table_new b
WHERE b.ds = '${ds}'
-- AND b.operation not in('D')
;
这个逻辑使用了一个JOIN加上一个UNION实现了一个MERGE逻辑，把增量合并成一份全量。这里有一个选项【-- AND b.operation not in('D')】，是否要把物理删除从当前全量表中删除，可以根据实际业务需求选择。
2.3.
业务计算逻辑
MERGE逻辑是最简单的一个涉及到增量的逻辑，但是实际业务计算逻辑要比这个场景更加复杂一些。
2.3.1.2张增量表的处理
我们在MERGE里面虽然也是2张表，但是其实这是一张表的增量与全量。如果是2张增量表，那么该如何处理呢。基于两张增量表无法关联的原则，我们必须引入全量表。
1.我们需要利用2张表的当日增量与全量，也就是说有4张表参与计算。
2.如果不想让全量直接关联，那么就需要先找到两个增量表的主键的并集。然后从两个表的全量中拆出这个并集的集合，再去关联。
逻辑如下：
-- ta_add ta表的增量表
-- ta_all ta表的全量表
-- tb_add tb表的增量表
-- tb_all tb表的全量表
-- 注意这个场景使用了mapjoin，增量表的数据量是有限制的
with tx_add as(
select distinct pk from(
select pk from ta_add where ds = '${ds}'
union all
select pk from tb_add where ds = '${ds}')t)
,ta_add2(
select /+mapjoin(t2)/ t1.*
from ta_all t1 join tx_add t2 on t1.pk=t2.pk
where t1.ds = '${ds}'
)
,tb_add2(
select /+mapjoin(t2)/ t1.*
from tb_all t1 join tx_add t2 on t1.pk=t2.pk
where t1.ds = '${ds}'
)
insert overwrite table tc_add partition (ds='${ds}')
select *
from ta_add2 t1 join tb_add2 t2 on t1.pk=t2.pk
;
这个逻辑利用了增量表比较小，可以利用了MAPJOIN的特性，可以快速的产出两个可以关联的并集再去关联。因为避免了大表的重分布，所以，可以大幅提升运行效率，降低资源消耗。（在这里增量的意义是表真的很大，如果全量是两张百万级的表，建议测试一下性能，可能直接关联更简单效率更高。所以，在MaxCompute做增量加工计算很多场景是没必要的。）
2.3.2.2张以上增量表的处理
我们一般说的增量加工的表还是指业务表，而不是代码表、参数表这种小表。这种万级的小表，增量与全量关联计算的性能差距可以忽略。百万级这种量级的表，增量计算也是意义不大的。我们看下上一小节那段冗长的逻辑，其实原本只需要2行就可以，现在已经变得如此的复杂。2张以上的表，如果使用同一PK关联，2张以上表的这个逻辑还是可以沿用的。如果有多个不同的关联PK，这个问题就从一维搞成了二维，除非实在不得已，不建议再去搞增量加工了。
我在这个优化工作的过程中遇到的场景，就是远远大于2张以上的表的增量加工，并且关联的PK也是多个。原来开发者选取了主表作为增量表，其他的表都是全量表的计算逻辑。因为这是一个分钟级的任务，原来的开发者应该还是希望从性能的角度做一些高效的设计。
但是在这个场景，第一主表并不是太大（百万级），相反左关联的表有上千万的。所以，我并未看到这个增量加工带来巨大性能提升的意义。第二主表的增量是通过一个指定的时间区间来识别近一个时间片段的，这种在集成的源是源系统的时候是可行的。但是这里刚好是一个不稳定的备库的备库，所以，使用固定时间区间可能会因为数据延迟导致未被识别为增量。
索性，我就直接改为全量加工了，这样就没问题了。但是这样就无法识别出哪些数据是加工都的增量了，这就涉及到下面要提到的增量推送的问题。
2.4.增量推送逻辑
有两种思路可以获取需要推送的增量，一种是从原始增量开始就一直保留增量标志字段，另一种是从最终结果中利用T和T+1两个全量比对出增量。在上面提到的场景，我们就遇到了第一个场景，我们需要在加工环节保持增量识别标志，并对这个字段在关联后的结果进行计算。
2.4.1.增量标志计算
增量标志要用来计算，一定是可以计算的。在上一节我提到系统有延迟业务的数据时间是不可靠的了，那如何来判断增量呢？我们的集成任务其实很难去从数据库的备库获取可靠的时间戳，但是我们本次集成的增量数据一定是一个确定的增量集合，所以，这个ETL_DATE（一般是我们dataworks的bizdate或者yyyymmddhhmiss）就是我们在MaxCompute批量做加工可用的增量时间戳。数据库同步工具识别出来的数据的变化状态有增、删、改、主键更新（I、D、U、K）四种，我们是可以直接利用的。
所以，我们在这里使用的逻辑如下:
select ...
,case when a.etl_partition ='${ds}' then a.etl_partition

when b.etl_partition ='${ds}' then b.etl_partition ... else a.etl_partition end as etl_date

【如何正确的做增量加工】,case when a.etl_partition ='${ds}' then a.operation

when b.etl_partition ='${ds}' then 'U' ... else a.operation end as operation

from tablea a
left join tableb on a.pk=b.pk
...
where ...;
所以这种方式是可以把增量状态保持下去的，但是因为这个计算后的结果其实一次次的叠加后，可能就不知道对不对了。所以，在具体的业务场景还要具体的去看。
2.4.2.全字段比对
全字段比对是一种暴力的计算方法，不需要增量加工，我也可以计算出增量。并且这种计算结果还是真实可靠的，相对于一个经过多层计算后的业务结果表来说，更是如此。
全字段比对逻辑如下：
一、T+1日表比T日表多的记录，INSERT状态；
二、T日表比T +1日表多的记录，DELETE状态；
三、T+1日表比T日表，关联后相同主键的非主键字段值不一致的，UPDATE状态；
这个比对十分消耗计算资源，尤其是一些最细业务粒度的交易表、事件表。但是对一些用户表这种表来说，问题倒是不大。比对逻辑如下：
-- I
select a.*,'I' as operation
from table1 a
left join table1 b on a.pk = b.pk and b.ds='${lastdate}'
where a.ds='${ds}'
and b.pk is null
;
-- D
select a.*,'D' as operation
from table1 a
left join table1 b on a.pk = b.pk and b.ds='${ds}'
where a.ds='${lastdate}'
and b.pk is null
;
-- U
select a.*,'U' as operation
from table1 a
join table1 b on a.pk = b.pk and b.ds='${ds}'
where a.ds='${lastdate}'
and(coalesce(b.col,'')<>coalesce(b.col,'') -- 字符
or coalesce(b.col,0)<>coalesce(b.col,0) -- 数值
or coalesce(b.col,'0001-01-01')<>coalesce(b.col,'0001-01-01')) -- 日期
;
全字段比对看起来其实并不优美，实在是有点粗暴。当然你也许会有更容易识别增量的方式，可以多试试，这将是你保底的方法。
3.实践总结
通过上面的内容，我们对增量加工的方法有了一定了解。希望我文中提到的方法能帮助大家在日后在项目中正确的使用增量加工的方法，并通过这个方法在部分场景获得显著的性能改进。另外我还是要提到一点，就是增量加工逻辑比全量加工更加复杂，并且还会遇到更为复杂的异常排查、补数据等维护等问题。大家在实际项目中，一定要权衡好利弊，再定夺方案。
原文链接：http://click.aliyun.com/m/100...
本文为阿里云原创内容，未经允许不得转载。