MaxCompute复杂数据分布的查询优化实践

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
简介:

2017年中国大数据技术大会于12月7-9日在北京新云南皇冠假日酒店隆重举行, 大会就大数据时代社会各行业的智能化进程和行业实践展开深入讨论。

在12月8日的“大数据分析与生态系统”分论坛上,来自阿里巴巴计算平台事业部的高级技术专家少杰,以“MaxCompute 复杂数据分布的查询优化实践”为题,为现场来宾分享了阿里云MaxCompute最新技术与实践的洞察与经验。
4

概述
数据分布的问题在大数据处理领域由来已久。很不幸,如今流行的大数据处理系统仍然没有很好地解决这个问题。在MaxCompute 2.0全新的优化器中,我们引入了复杂数据分布,添加了分区剪枝、分布上拉、下推以及分布对齐等优化措施。本文将从数据分布的历史和原理开始,介绍我们的思路和解决办法。

理解数据分布
提到数据分布,很多人会想到MPP DBMS。的确,我们通常说只有MPP DBMS才需要考虑数据分布优化。先考虑一个流行的分布式数据库分类学:

  1. Shared Everything: 区别于后两类,这一类基本不是分布式的。
  2. Shared Disk: 数据库服务器可以横向扩展,他们本身没有存储器,通过SAN或NAS技术连接到后端同样可以横向扩展的统一存储。受限于这层网络连接,数据库服务器的扩展能力非常有限。Oracle RAC等商业分布式数据库属于这类。
  3. Shared Nothing: 区别于Shared Disk,这种架构让数据库服务器和存储落在相同的物理节点上(co-located),使得物理节点之间不share任何信息,这大幅减少了网络IO。MPP DBMS和Hadoop属于这类。
    5

显然,只有Shared Nothing的数据库才需要考虑数据分布,你需要预知怎样把数据分布到不同的物理节点(而不是像Shared Disk那样放在统一存储),会使后续的操作代价更小。例如,在Greenplum中,必须在建表时指定partition key,系统会按照指定的key(哈希)分布数据。如果Join的两张表都按照join key来partition,这个Join就不需要网络IO。如果其中一张表使用了另一组partition key,那么可能要做一次re-partition。
这就是为什么要理解数据分布的原因:它对应用优化和系统优化都是非常重要的。MPP DBMS在数据分布上都有比较深的积累。但是为什么Hadoop这种大数据处理系统没有这类优化?是因为他们需要更强的扩展能力(以及非结构化数据支持,我们不展开这个话题)。
区别于MPP,Hadoop并不是在物理上强制数据和计算在相同节点,如果这么做,系统的横向扩展能力仍然受限。特别是动态扩展能力,考虑正在运行的一个50个节点的Greenplum集群,我们基本无法做到快速地加入例如2个节点还能高效工作。Hadoop在这方面是很在行的,它的解决办法主要是:
1、存储计算分离
2、去中心化的设计支持高效的peer to peer读写(HDFS)
这就是为什么你在Hive中创建一张表时,无须像Greenplum中那样指定partition key,同时Hive在Join的效率低于Greenplum的原因。

数据分布优化的目的
如上文所述,大数据分布式系统在存储系统上通常倾向随机分布,这提升了扩展性,牺牲了性能。但是重新审视这个权衡,在存储系统上随机分布并不意味着我们不能利用数据分布优化查询。分布优化的目的是希望尽可能的利用已经存在的分布,并尽可能满足未来要求的分布。这种优化包括:

1、分区剪枝:利用数据分布特性,我们可以做分区剪枝来减少数据读取。例如,哈希分布对于点查询,范围分布对于区间查询可以应用分区剪枝。
2、消除重分布:如果当前的分布满足后续算法的要求,我们可以消除额外的重分布操作。众所周知,重分布(在Hadoop中叫做shuffle)是分布式算法最主要的消耗。
3、避免数据倾斜:可以使用更好的数据分布算法避免数据倾斜。例如,某些单值重复率很高(end-biased)的数据集,使用范围分布而不是哈希分布可能会有效避免数据倾斜带来的性能影响。

定义
数据分布类型
数据分布类型和对应的意义和范例如下所示:
6
7

实现
在不破坏Volcano优化器语义的前提下,我们把分布特性实现为一种physical property,称作distribution。和其他property一样,它有required property和delivered property成对的属性。例如,对于sorted merge join,它对所有输入会施加一个Partial Ordered的required property,同时自身会deliver一个Partial Ordered property,这使得它的后继操作有机会利用这个property,避免一次重新分布。考虑以下查询:
11

此时Join如果被实现为Sorted Merge Join,它可能会deliver一个Hash[uid]的property,这正好被Aggregate要求,那么这里我们就可以省去一次不必要的重分布操作。
要做到类似的优化效果,我们需要关注下列问题:
1、收集分布特性
2、(局部关系代数编译)选择合适的分布特性
3、(全部代价计算上)规避不合适的分布特性
收集分布特性
产生数据分布有3种途径:
1、用户指定:就像MPP那样,可以在DDL中引入partition key,允许用户指定数据分布。当然区别于MPP,这种分布仅要求在分布式文件系统上的目录结构,并不能关联具体的物理节点。
2、SQL逻辑:SQL逻辑可能产生一次运行时的数据分布。例如distribute by字句声明了一次运行时的数据分布。
3、算法的副作用:每个分布式算法可能产生一次运行时数据分布。例如,sorted merge join可以保证它的输出数据满足按join key的有序和哈希分布的特征。

有若干算法要求一种特殊的数据分布:
1、Aggregate:Sorted Aggregate要求grouping key的Hash分布。
2、Join:Sorted Merge Join和Hash Join都要求输入按照join key的相同Hash分布。
3、Sort:Order by要求sort key上的Range分布,或Singleton分布。
选择合适的分布特性
即使给定了一系列required和delivered distribution property, 确定某个操作的分布仍然不是容易的事情。区别于ordering property(仅有排序列和升降序的属性),distribution property的变化很多,这些变化的原因包括:
1、满足要求的分布有多种选择。例如group by a, b, c这个aggregate,对输入有按a, b, c的Partial Ordered的要求,它对ordering的要求是a, b, c有序,但是满足它的分布可以是Hash(a), Hash(b), Hash(a,b), Hash(a,b,c), RNG(a)等不同的组合。
2、能利用的实现分布有多种选择。例如join a and b on a.id = b.id这个join,如果a服从Hashid, b服从Hashid,对于Sorted Merge Join,它可以选择要求Hashid,或Hashid,甚至任意Hash(id)。
这些复杂度加大了最优计划的搜索空间。事实上,最优计划是相对于关系代数数量的一个NPC问题。为了缩小搜索空间,我们引入了启发式的分支选择算法。在编译一个关系代数时,不仅需要满足后继操作的要求,还要考虑前序操作提供满足的分布的可能性,后者被实现为称作Pulled Up Property的模块。

12

Pulled Up Property猜测并筛选可能的前序delivered property,用于在编译时减少搜索宽度。考虑上图的查询,在Join编译时,因为Sink的需求下推,它被要求提供一个Hashc1。Pulled Up Property则从前序操作猜测可能会提供Hashc1和Hashc1,综合考虑,Join可能会直接要求Hashc1,从而减少了Hashc1和Hashc1这两个分支。

规避不合适的分布特性
数据倾斜(Skew)是指在分布中少量节点被分配了大部分数据,导致整个算法退化为单机操作。低并发(Under Partition)是指分布指定了过少的节点,是的分布式资源不能被有效利用。我们希望能避免这两种情况。
很显然,更好的统计信息会帮助我们规避这两种情况。Skew和Under Partition的情况下,需要对代价估计做相应的惩罚,降低他们被选为最优计划的可能性。我们定义”好”的分布是每个节点处理的数据量在一个预设的范围,低于或高于这个范围都会被施加惩罚。估计这个数据量的依据包括:
1、输入数据记录数(row count)
2、重复度最高的数据(top values)
3、直方图(histogram)

总结
在这篇文章中,我们介绍了数据分布优化的问题和意义,并解释了MaxCompute在数据分布优化上的实践。这一优化效果已经体现在MaxCompute最新的发布中。
从我们的测试来看,这个优化有相当显著的效果。我们对TPC-H进行了适当分区后,整体性能提升在20%的量级。即使没有对表数据分区,对用户完全透明的运行时分区优化也有很好的效果。在我们线上运行的环境中,14%的查询因为这个优化减少了至少一次数据重分布。

相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps 
相关文章
|
3月前
|
存储 SQL 数据管理
如何优化Hologres的性能?
【8月更文挑战第24天】如何优化Hologres的性能?
87 3
|
3月前
|
SQL DataWorks 数据库连接
实时数仓 Hologres操作报错合集之如何将物理表数据写入临时表
实时数仓Hologres是阿里云推出的一款高性能、实时分析的数据库服务,专为大数据分析和复杂查询场景设计。使用Hologres,企业能够打破传统数据仓库的延迟瓶颈,实现数据到决策的无缝衔接,加速业务创新和响应速度。以下是Hologres产品的一些典型使用场景合集。
|
3月前
|
SQL 分布式计算 数据安全/隐私保护
实时数仓 Hologres产品使用合集之重建表的索引后,如何将数据导入新表
实时数仓Hologres是阿里云推出的一款高性能、实时分析的数据库服务,专为大数据分析和复杂查询场景设计。使用Hologres,企业能够打破传统数据仓库的延迟瓶颈,实现数据到决策的无缝衔接,加速业务创新和响应速度。以下是Hologres产品的一些典型使用场景合集。
|
4月前
|
缓存 关系型数据库 MySQL
云原生数据仓库使用问题之如何查询表内数据分布的情况
阿里云AnalyticDB提供了全面的数据导入、查询分析、数据管理、运维监控等功能,并通过扩展功能支持与AI平台集成、跨地域复制与联邦查询等高级应用场景,为企业构建实时、高效、可扩展的数据仓库解决方案。以下是对AnalyticDB产品使用合集的概述,包括数据导入、查询分析、数据管理、运维监控、扩展功能等方面。
|
4月前
|
SQL Cloud Native 关系型数据库
云原生数据仓库使用问题之使用ORCA优化器时,怎么解决排序不准确的问题
阿里云AnalyticDB提供了全面的数据导入、查询分析、数据管理、运维监控等功能,并通过扩展功能支持与AI平台集成、跨地域复制与联邦查询等高级应用场景,为企业构建实时、高效、可扩展的数据仓库解决方案。以下是对AnalyticDB产品使用合集的概述,包括数据导入、查询分析、数据管理、运维监控、扩展功能等方面。
|
5月前
|
机器学习/深度学习 分布式计算 大数据
MaxCompute产品使用问题之如何优化大数据量的查询和处理
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
5月前
|
SQL 分布式计算 DataWorks
MaxCompute产品使用问题之如何同时写入多表
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
6月前
|
存储 分布式计算 安全
MaxCompute的性能
【5月更文挑战第6天】MaxCompute的性能
78 2
|
存储 并行计算 Cloud Native
PolarDB 开源版通过 brin 实现千分之一的存储空间, 高效率检索时序数据
PolarDB 的云原生存算分离架构, 具备低廉的数据存储、高效扩展弹性、高速多机并行计算能力、高速数据搜索和处理; PolarDB与计算算法结合, 将实现双剑合璧, 推动业务数据的 价值产出, 将数据变成生产力. 本文将介绍PolarDB 开源版通过 brin 实现千分之一的存储空间, 高效率检索时序数据
208 0
|
6月前
|
关系型数据库 分布式数据库 PolarDB
PolarDB 开源版通过 brin 实现千分之一的存储空间, 高效率检索时序数据
背景PolarDB 的云原生存算分离架构, 具备低廉的数据存储、高效扩展弹性、高速多机并行计算能力、高速数据搜索和处理; PolarDB与计算算法结合, 将实现双剑合璧, 推动业务数据的 价值产出, 将数据变成生产力.本文将介绍PolarDB 开源版通过 brin 实现千分之一的存储空间, 高效率检...
127 0