推荐：SQL支持度和粒度已经成为当今所有大数据计算引擎的重要指标之一，SQL的所有操作都可以分解为简单的操作（过滤操作如如where、limit等）和复杂操作（groupby、join等聚合操作），而join操作是其中最复杂、开销最大的操作类型，是大多数业务场景的性能瓶颈。因此，今天我们以dorissql为基础，简单说说SQL支持的几种常见的join算法及其适用场景，以了解在设计和优化计算引擎时需要关注和考虑的问题。基于此你也可以掌握在大数据场景下做业务开发时，需要什么样的优化策略来适应你的数据量和逻辑。虽然我们这里说的是dorissql，但是对于使用spark、hive等其他计算引擎来说还是很有用的。这也是类似的道理。好吧，我们开始吧，玩得开心！——飓风飓风，MobTech大数据研发总监由世说网与上海白玉兰开源与开放研究院联合举办的开源大数据技术在线Meetup如期举行。ApacheDoris社区受邀参加了本次Meetup。百度高级内核研发工程师、ApacheDorisContributor李浩鹏数据为大家带来了《ApacheDorisJoin实现与调优实践》的专题分享，主要介绍了ApacheDorisJoin的实现机制和实用的调优策略。以下为分享。很高兴参加这次开源大数据技术交流会。今天和大家分享的话题是ApacheDorisJoin的实现和调优。内容主要分为三部分：第一部分是给对ApacheDoris不太了解的人。简单介绍一下Doris，第二部分会介绍Doris整个Join的实现机制，第三部分我们将如何基于Doris的这些Join实现机制来进行Join的调优工作。Doris简介首先，简单介绍一下Doris。Doris是百度自主研发并开源的基于MPP（MassivelyParallelProcessing）架构的分析型数据库。其特点是性能卓越，可实现毫秒级/秒级响应，进行PB级数据分析。适用于低延迟下的实时报表、多维分析等高并发需求场景。多丽丝原名帕洛。2017年我们以百度Palo的形式开源在GitHub上。2018年，我们将其贡献给Apache社区，并正式更名为ApacheDoris。在百度中，已经使用了Palo这个名字，在百度智能云上提供了Palo的企业级托管版本。Doris的历史始于2008年，最早应用于百度丰巢统计报表的场景。2009年，我们将其推广，开始承接百度内部的其他报表业务。2012年，我们全面升级了Doris的性能、易用性和可扩展性，基本承担了百度内部所有的统计报表业务。2013年，我们升级了Doris的MPP框架，开始支持分布式计算。2015年系统架构大幅简化，主要架构沿用至今。自百度Palo于2017年正式开源并于2018年贡献给Apache社区以来，截至目前，ApacheDoris在Github上的Star数约为2.9k+，在美团、小米、京东、网易、字节跳动的贡献者约180+,快手等多家一线互联网公司广泛使用。下图是Doris在整个数据分析场景中的定位。Doris自己承担了数据分析的角色。从传统的数据源，包括RDMS，到业务应用，包括一些WEB和移动端的日志，数据快速接入到InDoris，同时赋能各种数据应用，提供数据分析服务。Doris是一个基于MPP架构的分析型数据库。它有几个特点：第一个特点是简单易用，支持标准SQL，完全兼容MySQL协议。该产品使用起来非常方便。其次，它使用了预聚合技术、矢量化执行引擎和列式存储。是一个高效的查询引擎，可以在秒级甚至毫秒级返回海量数据下的查询结果。第三，它的架构非常简单，只有两组进程：FE负责管理元数据，解析SQL，生成和调度查询计划；BE负责存储数据和执行FE生成的查询计划。这种简单高效的架构使其易于运维，易于部署，扩展性强，能够支持大规模计算。通过下图，我们简单梳理一下Doris的结构。Doris主要分为两个角色，一个是FE，一个是BE。从SQL执行的角度来看，FE在Doris中承担了MySQL的访问层，负责查询计划的解析、生成和调度。BE负责执行相应的查询计划，并负责实际的查询、导入等工作。从数据的角度来看，FE负责元数据的存储，比如表、数据库、用户信息等数据，BE负责列存数据的落地存储。这个架构很简单，每个BE节点都是平等的。FE分为几个角色：Leader、Follower、Observer。这类似于ZooKeeper中的角色定位。Leader和Follower参与集群Master的选择和元数据的修改，而Observer不参与这个过程。只提供数据读取，对外提供FE读取扩展性，因此FE和BE节点都可以线性扩展。接下来是Doris中数据的分布式存储机制。Doris作为一个MPP数据库，其数据存储方式将深刻影响我们后面分析的Join的实现和优化。Doris可以支持多副本存储，数据可以自动迁移，实现副本平衡。我们看到Doris中的数据是以tablet的形式组织的，每张表会被拆分成多个tablet，每个tablet由数据分区和数据桶决定。一旦确定了tablet，Doris中的所有数据都会根据tablet进行调度。我们可以看到一个tablet可以存储在多个BE上，实现多副本存储。如果一个BE节点宕机或者有新的BE节点加入，系统会在后台自动平衡数据副本。查询的时候，查询负载也会均衡到所有的BE上。这是Doris在数据副本存储上的整体架构。后面我们做Join分析的时候，也会看到数据的副本，包括数据在其中是怎么调度的。的。DorisJoin实现机制Doris支持两种物理算子，一种是HashJoin，一种是NestLoopJoin。HashJoin：在右表上基于等价的Join列建立哈希表，左表使用哈希表以流式的方式进行Join计算。它的局限性在于只能应用于等价的Join。NestLoopJoin：通过两个for循环，很直观。那么适用于不等值Join的场景，比如：大于小于或者需要笛卡尔积的场景。它是一个通用的Join运算符，但性能较差。作为分布式MPP数据库，Join过程中需要进行数据Shuffle。需要对数据进行拆分和调度，以保证最终的Join结果是正确的。举个简单的例子，假设关系S和R是join的，N表示参与join计算的节点数；T表示关系中元组的数量。Doris支持4种数据混洗方式：BroadCastJoin，需要将右表的所有数据发送到左表，即每个参与Join的节点都有右表的全量数据，即T（右）。它的适用场景比较通用，可以同时支持HashJoin和NestloopJoin，网络开销为N*T(R)。ShuffleJoin在进行HashJoin时，可以通过Join列计算出对应的Hash值，进行Hashbucketing。它的网络开销是：T(R)+T(N)，但是它只能支持HashJoin，因为它也是根据Join的条件计算buckets。BucketShuffleJoinDoris的表数据是通过Hash计算分桶的，所以可以利用表本身的桶列的性质对Join数据进行shuffle。如果两张表需要join，并且join列是左表的bucket列，那么实际上可以计算左表的数据，而不用移动右表的数据，通过数据的bucket发送数据在左表中。它的网络开销为：T(R)相当于只对右表的数据进行Shuffle。Colocation类似于BucketShuffleJoin，意思是在导入数据的时候，已经按照预设的Join列的场景做了数据Shuffle。那么在实际查询时，可以直接进行join计算，无需考虑datashuffle问题。下图是BroadCastJoin。左表中的数据没有移动。右表中每个BE节点扫描到的数据发送给对应的Join节点。每个Join计算节点都有右表中的全量数据。第二个是ShuffleJoin。每个数据扫描节点扫描数据后进行Partition分区，然后根据Partition分区的结果将左右表的数据发送给对应的Join计算节点。下图是BroadCastJoin。左表中的数据没有移动。右表中每个BE节点扫描到的数据发送给对应的Join节点。每个Join计算节点都有右表中的全量数据。第二个是ShuffleJoin。每个数据扫描节点扫描数据后进行Partition分区，然后根据Partition分区的结果将左右表的数据发送给对应的Join计算节点。第三张图是BucketShuffleJoin。右表数据扫描完成后，进行数据分区的Hash计算，将左表自身的数据分布发送给对应的Join计算节点。最后一个是CoLocateJoin。其实它并没有真正的datashuffle，数据扫描后进行join计算就OK了。以上四种方法的灵活性从高到低。它对数据分布的要求越来越严格，但是Join计算的性能越来越好。接下来分享的是Doris最近新增的一个新特性——RuntimeFilter的实现逻辑。Doris在进行HashJoin计算时会在右表上构建哈希表，左表流经右表上的哈希表得到Join结果。RuntimeFilter充分利用了右表的Hash表。当右表生成哈希表时，同时生成基于哈希表数据的过滤条件，然后下推到左表的数据扫描节点。这样，Doris就可以在运行时进行数据过滤。如果左表是大表，右表是小表，那么在Join层需要过滤的大部分数据，在读取数据的时候，可以通过左表生成的过滤条件提前过滤。提高连接查询的性能。目前Doris支持三种类型的RuntimeFilter。一种是IN——IN，很好理解，把一个hashset下推到数据扫描节点。第二种是BloomFilter，利用哈希表中的数据构造BloomFilter，然后将BloomFilter下推到扫描节点进行数据查询。最后一个是MinMax，就是一个Range区间。通过右表数据确定Range范围后，下推到数据扫描节点。RuntimeFilter的适用场景有两个要求：第一个要求是右表大，左表小，因为构建RuntimeFilter需要计算成本，包括一些内存开销。第二个需求是左右表join的结果很少，说明这个join可以过滤掉左表的大部分数据。当满足以上两个条件时，启用RuntimeFilter会产生更好的效果。当Join列是左表的Key列时，RuntimeFilter会被下推到存储引擎。Doris本身支持延迟物化。延迟物化简单来说就是这样：如果需要扫描ABC的三列，A列有一个过滤条件：A等于2，如果要扫描100行，可以先扫描A列的100行来out，然后通过A=2的过滤条件进行过滤，过滤完成后的结果，读取BC列，可以大大减少数据读取IO。所以如果在Key列上生成RuntimeFilter，同时利用Doris自身的延迟物化进一步提升查询的性能。让我们简要比较三种不同类型的运行时过滤器。IN的优点是效果滤波效果明显，速度快。它的缺点一是只适用于BroadCast，二是当右表超过一定数据量时会失效。目前Doris当前配置的是1024，即如果右表大于1024，IN的RuntimeFilter会直接失效。MinMax的优点是开销比较小。它的缺点是对数值列效果较好，但对非数值列基本没有效果。BloomFilter的特点是通用，适合各种类型，效果比较好。缺点是其配置较复杂，计算量较高。最后，DorisJoin还有一个重要的机制——JoinReorder，在进行Join调优的时候经常用到。一旦数据库涉及多表Join，Join的顺序对整个Join查询的性能影响很大。假设有3个表join，参考下图，左边是a表和b表的join，中间结果有2000行，再和c表join计算。接下来看右图，调整Join的顺序。先把a表和c表join，生成的中间结果只有100，最后再和b表join计算。最后的Join结果是一样的，但是它产生的中间结果有20次的差距，会产生很大的性能Diff。Doris目前支持基于规则的JoinReorder算法。它的逻辑是：让大表和小表尽可能join，它产生的中间结果尽可能小。把conditionalJoin表往前放，也就是尽量过滤conditionalJoin表。HashJoin的优先级高于NestLoopJoin，因为HashJoin本身比NestLoopJoin快很多。DorisJoin调音练习接下来进入第三部分，DorisJoin调音练习。在第二部分分享完Join机制后，第三部分需要使用到Doris自身的一些Join特性，包括Doris提供的Join调优机制。下面是DorisJoin的调优方法：使用Doris自己提供的Profile来定位查询的瓶颈。Profile会记录Doris整个查询过程中的各种信息，这些信息是性能调优的第一手资料。理解Doris的Join机制，这也是第二部分给大家分享的内容。只有知其然，知其所以然，了解其机理，才能分析出为什么会慢。使用Session变量改变Join的一些行为，实现Join的调优。查看QueryPlan分析这个调优是否生效。以上4步基本完成了一个标准的Join调参流程，接下来就是实际查询验证，看看效果如何。如果把前面4种方法串联起来，还是不行。这时候可能需要重写Join语句，或者调整数据分布。需要重新检查整个数据分布是否合理，包括查询Join语句。可能需要进行一些手动调整。当然，这种方法的心理成本比较高，也就是说，只有在尝试前面的方法都不行的情况下，才需要做进一步的分析。下面通过几个实际案例来分享Join的分析和优化过程。看下图中的Profile，四张表的一个join查询。在通过Profile的时候，发现第二次Join耗时很长，需要14秒。Profile进一步分析，发现BuildRows，即右表的数据量约为2500万。而ProbeRows（ProbeRows是左表的数据量）只有10000多条。在这种场景下，右表比左表大很多，这显然是一种不合理的情况。这明显说明Join的顺序有问题。这时候尝试改变Session变量，开启JoinReorder。setenable_cost_based_join_reorder=true这次耗时从14秒降到了4秒，性能提升了3倍多。此时再次查看profile时，左右表的顺序已经调整正确，即右表为大表，左表为小表。基于小表构建哈希表的开销非常小。这是一个典型的场景，其中使用JoinReorder来提高Join性能。在第二种情况下，查询速度很慢。查看Profile后，整个Join节点大约需要44秒。它的右表有1000万，左表有6000万，返回的结果只有6000万。这里可以大致估计过滤率很高，那么为什么RuntimeFilter没有生效呢？通过QueryPlan查看，发现只启用了IN的RuntimeFilter。前面说过，当右表超过1024行时，IN不会生效，所以根本起不到过滤作用，所以尽量调整RuntimeFilter的类型。这里改成BloomFilter，过滤左表6000万条数据中的5900万条。基本上过滤掉了99%的数据，这个效果是显着的。查询时间也从原来的44秒降到了13秒，性能提升了三倍以上。下面是一个极端的情况，通过一些环境变量的调优是没有办??法解决的，因为涉及到SQLRewrite，所以这里列出了原来的SQL。这个Join查询很简单，就是左右表的Join。当然，这上面还有一些过滤条件。打开Profile，发现整个查询HashJoin执行了三分多钟。它是一个BroadCastJoin，它的右表有2亿条条目，而左表只有70万条。这种情况下选择BroadcastJoin是不合理的，相当于做了一个2亿表项的HashTable，然后用70万表项遍历2亿个HashTable，显然是不合理的。为什么会出现不合理的Join顺序？左表其实是一张10亿条的大表，在里面加了两个过滤条件。加上这两个过滤条件后，10亿个条目只有70万个条目。但是Doris目前并没有一个很好的统计信息收集的框架，所以不知道这个过滤条件的过滤率是多少。所以在安排join顺序的时候，join的左右表选择顺序错误，导致性能极低。下图是改写完成后的SQL语句。在Join后面加一个JoinHint，在Join后面加一个方括号，然后写需要的Join方法。这里选择了ShuffleJoin，可以看到右边实际查询计划中的数据确实是分区的。原来的3分钟耗时改写后，只剩下7秒了，性能明显提升。接下来，我将根据今天分享的内容，对最佳实践原则做一个总结。主要分为4点：第一点：在做Join的时候尽量选择同类型或者简单类型的列。如果同一个类型是同一个类型，减少它的数据Cast，简单类型本身的计算就非常快。第二点：Join尽量选择Key列。原因在前面的RuntimeFilter中也有介绍。Key列可以对延迟实现有更好的效果。第三点：大表之间的Join，尽量做成Co-location，因为大表之间的网络开销非常大，如果需要做Shuffle，成本非常高。第四点：合理使用RuntimeFilter，在Join过滤率高的场景下非常有效。但它不是万能的，而是有一定的副作用，所以需要根据具体SQL的粒度进行切换。最后：说到多表join，需要判断join的合理性。尽量保证左表是大表，右表是小表，那么HashJoin会比NestLoopJoin好。如果需要，您可以使用Hint通过SQLRewrite调整Join的顺序。后续规划最后，要给大家介绍一件比较精彩的事情。今天分享了很多Join调优实践的一些方法论，但是很多Join需要调优的原因是Doris的自动调优不够好。社区的一个重点工作方向是更智能的优化器，它可以大大降低手动调优的心理成本。那么更好的智能优化器取决于什么？是一个更全面的统计集合。因此，社区会在Q3做更详细的统计信息收集，进而进一步提高查询优化器的自动化程度。希望下个版本发布的时候，大家可以忘掉今天和大家分享的内容。这是我们希望Doris将来可以做的事情。最后，在百度智能云上，百度提供了基于ApacheDoris快速迭代的企业级托管版本。有兴趣的小伙伴可以试试。谢谢大家的聆听，谢谢大家。
