如何让数据增量更新节省资源,耗时减半

内容来源:本文内容由阿里大数据计算服务(MaxCompute)团队投稿提供。IT 大咖说(微信id:itdakashuo)经授权发布,转载请标明出处。

阅读字数:5502 | 14分钟阅读

摘要

目前hash clustering table已经在阿里巴巴内部集群生产环境正式发布,并且已经有蚂蚁、安全部、菜鸟等多个BU参与了试用。从蚂蚁的反馈来看,改造之后的任务收效非常明显,运行时间缩短40%到80%,节省计算资源23%到67%。

对于增量更新的场景,可以利用 MaxCompute2.0的新特性,对语句做简单改造,从而大幅提升性能,节约集群资源。

背景介绍

在数据开发的过程中,往往会进行分层的设计,在ODS层中,一种非常常见的场景是使用一个增量表delta对一个存量表snapshot进行更新。例如snapshot表存储所有的会员信息,而增量表中包括新增会员信息和原有会员信息属性的一些修改;或者snapshot表存储最近一个月的订单信息,delta表存储了新增订单以及物流的更新等等。

对于这种任务,往往有以下几个特征:

  1. snapshot表存储量巨大,delta表相对较小
  2. snapshot表和delta表拥有一致的schema
  3. snapshot和delta表中存在主键key,且key可能有重合(否则可以通过简单的union all来完成)
  4. 上一个周期的snapshot + 当前周期的delta => 当前周期的snapshot

为了完成上述的功能,对应的sql逻辑一般使用full outer join,简单起见,我们的snapshot和delta表只有两列

(key string, value string)

其中key为主键

INSERT OVERWRITE TABLE snapshot PARTITION (ds='20170102')
SELECT

CASE WHEN d.key IS NULL THEN s.key ELSE d.key END,
CASE WHEN d.key IS NULL THEN s.value ELSE d.value END

FROM
(SELECT * FROM snapshot WHERE ds='20170101') s
FULL OUTER JOIN
(SELECT * FROM delta WHERE ds = '20170102')d
ON s.key = d.key;

这个语句表示,对于delta表存在的数据,使用delta表的值,对于delta表不存在的数据,使用snapshot的值。

问题

但是在实际执行的过程中,虽然只是进行简单的join操作,但是由于存量表可能非常大(几T到几十T的规模),这种任务往往耗时非常长,有些任务甚至需要耗费一天的时间才能产出。这种任务是否存在优化的空间呢?我们可以分析一个线上实际的大表full outer join的执行计划。

可以看到M1是snapshot表,需要将近20000的并发,M2是delta表,只有9个并发,而为了进行join的操作,两边会各自进行shuffle,在J3阶段进行sort-merge-join的计算。实际执行过程中,M2只需要几分钟,M1需要十几分钟,而在J3阶段则往往需要一两个小时,因为J3只有3000个并发,但是却读取了上游将近20000个并发读取的数据,相当于并发减小到原来的15%,处理的数据量却是一样的,当然耗时会长。另外,从M1到J3这个路径的shuffle中间存在大表的一次读写+两次排序,而且在数据量较大的情况下,还有可能会发生数据spill,使得运行性能更差。

在这种情况下,为了缩短执行时间,通常可以调大join阶段的instance数目,增加join阶段的内存减少spill等,但是instance的数目不能无限增长,否则会由于shuffle规模太大造成集群压力过大,另外内存的资源也是有限的,所以调整参数也只是牺牲资源换取时间,治标不治本。

为了对这个场景进行彻底的优化,我们希望能完全消除掉大表的shuffle阶段,将M1和J3合二为一,这样大表数据只需要读写一次,而且免去了中间排序的过程,执行时间可以缩短一半甚至更多。有调优经验的同学可能已经想到了mapjoin,但是这里的delta表往往数据较多无法当做mapjoin的小表,另外mapjoin无法支持full outer join,这两个限制都无法绕过,所以这个方案只能被pass了。那么这个shuffle的阶段应该如何省去呢?这里就要引入我们今天介绍的功能,hash clustering table了。

方案

Hash clustering,简而言之,就是将数据提前进行shuffle和排序,在使用数据的过程中,读取数据后直接参与计算。这种模式非常适合产出后后续节点多次按照相同key进行join或者聚合的场景。当然生成hash clustering table本身也是有代价的,在生成阶段会进行一次额外的shuffle。因此,这个功能并不是对于所有的场景都有效,例如数据生成之后只使用了一次,那么这个shuffle在生成表的阶段进行还是在读表之后进行其实并没有什么区别。但是对于特定的场景,这个特性可以起到显著的效果。

根据这个方案,我们重建一下snapshot表

ALTER TABLE snapshot CLUSTERED BY (key) SORTED BY (key) INTO 100 BUCKETS;

注意这个100 bucket需要根据实际数据规模进行设置,这里只是示例,不要照抄^_^

然后重建一下ds='20170101'的数据

INSERT OVERWRITE TABLE snapshot PARTITION (ds='20170101')
SELECT key, value
FROM snapshot
WHERE ds='20170101'

注意,这个过程由于会有一个额外的shuffle阶段,所以耗时会比普通的insert overwrite长。

  • 第一次尝试: full outer join

数据准备完成后,重新执行刚才的full outer join语句

INSERT OVERWRITE TABLE snapshot PARTITION (ds='20170102')
SELECT

CASE WHEN d.key IS NULL THEN s.key ELSE d.key END,
CASE WHEN d.key IS NULL THEN s.value ELSE d.value END

FROM
(SELECT * FROM snapshot WHERE ds='20170101') s
FULL OUTER JOIN
(SELECT * FROM delta WHERE ds = '20170102')d
ON s.key = d.key;

让我们看下执行计划

结果好像不尽如人意,M1读取了delta表,M2读取了snapshot表并且进行了sort-merge-join操作,但是读取完成以后数据重新进行了一次shuffle才写入了ds='20170102'分区,为什么会这样呢?

原因是ds='20170102'这个分区也是一个hash clustering table的分区,在写入的过程中,也需要数据按照特定key进行shuffle,虽然ds='20170101'的数据是shuffle过了的,但是在后续的full outer join的过程中,可能会存在补null的行为,并不能保证输出数据依然符合shuffle的特征,所以需要进行一次reshuffle。

其实,这个sql通过CASE WHEN d.id IS NULL THEN s.id ELSE d.id END在语义上实际是保证了不会出现额外补null的行为的,但是这个行为目前我们的优化器还不能识别,所以这种情况下大表数据依然会有一次shuffle,这并不能让我们满意。

  • 第二次尝试: not in + union all

下一个问题是如何才能让优化器识别出来我们其实并没有改变shuffle的属性呢,我们观察到这个full outer join其实这个sql就是一个求并集的过程

那么整个这个sql可以被拆分为两部分

SELECT a.key, a.value
FROM (SELECT * FROM snapshot WHERE ds='20170101' AND KEY NOT IN

(SELECT  key FROM delta WHERE ds='20170102')) a  -- snapshot_not_in_delta

UNION ALL
SELECT key, value FROM delta WHERE ds='20170102' -- delta_all

在上述两部分中,前一部分对应图中的蓝色部分,后一部分对应图中的绿色部分。我们仅仅是对snapshot的key列进行了过滤操作,并没有改变key的分布,所以这个语句可以省去一次额外的shuffle。但是MaxCompute对于not in有一个限制是结果集合不能超过2000条,这个又限制了这种写法的应用场景。

  • 最终方案: anti semi join + union all

好在MaxCompute2.0中新支持的anti semi join同样实现了not in的语义,而且对结果集大小并没有限制,使用anti semi join 这个语句可以进一步修改为

INSERT OVERWRITE TABLE snapshot PARTITION (ds='20170102')
SELECT s.key, s.value 
FROM (SELECT * FROM snapshot WHERE ds='20170101') s 
LEFT ANTI JOIN
(SELECT * FROM delta WHERE ds='20170102') d ON s.key = d.key
UNION ALL
SELECT key, value
FROM delta
WHERE ds='20170102';

经过这一步的改造后,让我们运行一下,看看发生了什么。

只有三个阶段,M1读取delta表,M2读取snapshot表并进行sort-merge-join,随后写出数据,最后一个R3阶段仅仅是一个收集信息的任务,耗时在秒级别,所以实际的处理阶段只有两个stage,其中M1合并了之前M1和J3的功能,由于省去了一次数据读写、排序以及可能的spill等操作,实际运行时间往往可以减半。

上面也说过,如果数据只是进行一次读写,其实hash clustering table的作用有限,但是在增量更新这个特定的场景下,我们的输入和输出都为hash clustering的数据,而且中间过程并没有对cluster key进行修改,只是进行了过滤,所以我们可以只在一个阶段中完成read->join->union all->write这四个操作,极大地缩短了运行时间。

收效

目前hash clustering table已经在阿里巴巴内部集群生产环境正式发布,并且已经有蚂蚁、安全部、菜鸟等多个BU参与了试用。

从蚂蚁的反馈来看,改造之后的任务收效非常明显,运行时间缩短40%到80%,节省计算资源23%到67%。

菜鸟在使用hash clustering之后,任务的执行计划有所变化,节省了之前join操作需要的shuffle等操作,任务执行时间从40分钟左右降低到20分钟以内,有效的提升了任务执行效率,缩短执行时间,节约了资源。

飞猪应用Hash Clustering后,对于计算,整个计算过程由优化前的3小时,缩短到40分钟内完成,对于明细事实表视图一次读取计算可在1分钟内完成;对于存储,节省的存储和数据膨胀程度是线性关系,采用视图形式,我们用非常小的计算消耗代价节省了80%的存储,这一点看来,是很值得的。

所以我们付出的代价,仅仅是将表的属性进行修改,并且提前进行一次数据生成操作,这个操作也只需要执行一次,一劳永逸。

最后,欢迎大家在自己的增量更新的任务使用hash clustering功能,从现有的经验来看,大表的数据越多,收益越明显。

一些需要注意的地方

  1. bucket的数目设置需要一些经验,bucket越多,并发越多,运行越快,但是如果文件本身不大,小文件也越多,目前推荐500MB~1GB设置一个bucket,超大规模数据情况下一个bucket的数据可以更多。在任何情况下,不建议设置bucket number超过4096。
  2. hash clustering table会对数据进行重排,在一些极端场景下,可能会导致原来压缩率较高的文件压缩率降低,影响后续的性能,这个可以通过观察生成表的summary的input/output bytes来确认
  3. 目前我们正在对decimal类型进行重构,重构之后可能会影响decimal类型的分布方式,所以clustered key不要选用decimal类型
  4. snapshot表和delta表的schema不需要完全一致,但是如果key的类型不同,比如一边是bigint,一边是string,在join的时候需要将delta表的类型转换为snapshot的key类型,否则依然会需要一次reshuffle。

原文发布于微信公众号 - IT大咖说(itdakashuo)

原文发表时间:2018-11-13

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏数据和云

DBA入门之路:察微知渐细致入微

在DBA的职业生涯中,要面临无数的艰难险阻、排忧解难,所以细致入微,严谨认真的风格必不可少。养成了察微知渐的习惯,才能在分析诊断故障时层剖缕析,直指核心;而我也...

2123
来自专栏杨建荣的学习笔记

merge语句导致的性能问题紧急优化 (r9笔记第85天)

晚上正在休息的时候,突然收到一封报警邮件。 报警内容: CPU utilization is too high ------------------------...

2815
来自专栏IT派

SQL的巨大飞跃:MySQL 8.0发布

“你仍在使用SQL-92吗?”是我在“新SQL”演讲中的开篇问题。在我提出这个问题后,竟然有大部分观众坦承仍在使用25年前的技术。而如果我问谁还在使用Windo...

1504
来自专栏杨建荣的学习笔记

用Oracle的眼光来学习MySQL 5.7的sys(下)(r11笔记第25天)

昨天写了篇分析sys的文章,用Oracle的眼光来学习MySQL 5.7的sys(上)(r11笔记第24天) 收到了一些朋友的反馈,还不错,今天继续努力,再整理...

36313
来自专栏Python爬虫实战

MySQL从零开始:数据库简介

从字面意思理解,首先数据库是一个存放东西的库,里面存的东西是数据。以下解释来自百度百科:

1312
来自专栏bboysoul

给自己的主机测个速度

昨天有人问我,为什么他的国外服务器看有土鳖的视频那么卡,其实归纳一下有下面这么几点导致你的服务器速度不高

1062
来自专栏数据和云

Oracle智能之SQL诊断:SQL Tuning Advisor推荐执行计划

编辑手记:在前一段,一篇智能数据库优化的论文引起广泛的关注,其实在 Oracle 数据库中,已经引入了大量自动化和智能化的方法去进行自动调节,包括在 SQL 层...

3005
来自专栏数据和云

SQL审核:OR展开与子查询优化案例详解

黄廷忠(网名:认真就输) 云和恩墨技术专家 个人博客:http://www.htz.pw/ 本篇整理内容是黄廷忠在“云和恩墨大讲堂”微信分享中的讲解案例,S...

3684
来自专栏乐沙弥的世界

收缩Oracle数据文件

      最近有网友提到收缩Oracle数据文件的问题,这是DBA经常碰到的一个常见问题。通常我们需要收缩相应的数据文件以减少来自磁盘空间的压力以及提高数据库...

1072
来自专栏Linyb极客之路

一位Java工程师的阶段性工作总结

1.1.1、通常的模块分布:一般如果你要实现一个web应用,你从后台将数据展示到前端页面,在一个比较大的公司,你少不了跟其他项目有交集(你调用他的接口,他依赖你...

1033

扫码关注云+社区

领取腾讯云代金券