举荐语：
SQL 的反对力度和粒度，曾经作为明天所有大数据计算引擎的重要衡量标准之一，而 SQL 的所有操作，能够分为简略操作（如 where、limit 等 filter 操作）和简单操作（groupby、join 等聚合操作），而 join 操作又是其中最简单、代价最大的操作类型，是大部分业务场景的性能瓶颈所在，所以，明天咱们基于 doris sql，来简要的聊一下 SQL 所反对的几种常见的 join 算法以及其实用场景，管中窥豹，以此来理解设计和优化一个计算引擎的时候，哪些是须要重点关注和考量的，基于此也能把握平时做大数据场景的业务开发时，须要什么样的优化策略去适配你的什么样的数据量和逻辑，尽管这里聊的是 doris sql，但对应用 spark、hive 等其余计算引擎也是类似的原理。好了，咱们开始了，have fun！——MobTech 袤博科技大数据研发总监
飓风
由示说网和上海白玉兰开源凋谢研究院联结举办的开源大数据技术线上 Meetup 如期举行，Apache Doris 社区受邀参加本次 Meetup，来自百度的数据内核高级研发工程师、Apache Doris Contributor 李昊鹏为大家带来了题为“Apache Doris 的 Join 实现与调优实际”的主题分享，次要介绍了 Apache Doris Join 的实现机制以及调优策略实战，以下是分享内容。
非常高兴能够参加本次的开源大数据技术 Meetup，明天跟大家分享的主题是 Apache Doris 的 Join 实现和调优，内容次要分为三块：第一局部会先给不太理解 Apache Doris 的小伙伴们简略介绍一下 Doris，第二局部会介绍 Doris 的整个 Join 实现的机制，第三局部是咱们基于 Doris 这些 Join 实现机制将怎么开展 Join 的调优工作。
Doris 简介
首先简略介绍一下 Doris。Doris 是百度自主研发并开源的一个基于 MPP（大规模并行处理）架构的剖析型数据库，它的特点就是性能卓越，可能做到 PB 级别的数据分析的毫秒 / 秒级的响应，实用于高并发低延时下的实时报表、多维分析等需要场景。Doris 最早是叫 Palo，2017 年咱们以百度 Palo 的形式在 GitHub 上进行了开源，在 2018 年的时候把它奉献给 Apache 社区，正式更名为 Apache Doris。而在百度外部始终沿用了 Palo 的名称，并在百度智能云上提供了 Palo 的企业级托管版本。
Doris 的历史是从 2008 年开始的，最早是利用于百度凤巢统计报表的场景。在 2009 年咱们对它进行了通用化革新，开始承接百度外部的其余报表业务。在 2012 年咱们对 Doris 的性能、可用性以及扩展性进行了全面的降级，基本上承当起百度外部的所有统计报表的业务。
在 2013 年咱们对 Doris 做了 MPP 框架的降级，开始反对了分布式计算。在 2015 年对系统架构进行了大幅精简，主体架构始终沿用至今。从 2017 年百度 Palo 正式开源并于 2018 年奉献给 Apache 社区，截止目前，Apache Doris 在 Github 上有大略 2.9k+ Stars，Contributors 大略有 180+，在美团、小米、京东、网易、字节、快手等泛滥一线的互联网公司当中都有宽泛的应用。
接下来这张图是 Doris 在整个数据分析场景当中的定位。Doris 它自身承当的是一个数据分析的角色，从传统的数据源，包含 RDMS、业务利用，包含一些 WEB 端、挪动端的日志，通过 Spark、Flink 等批计算或流计算引擎将数据疾速接入到 Doris 当中，并对外向各类数据利用赋能、提供数据分析的服务。
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 参加到集群选主、元数据的批改等工作，而 Observer 是不参加这个过程的，只提供数据的读取，对外提供 FE 的读扩展性，所以 FE 与 BE 节点都能够线性的扩大。
接下来是 Doris 当中数据的分布式存储机制，Doris 作为一个 MPP 数据库，它的数据存储会深刻影响到后续咱们要剖析的 Join 实现与调优。
Doris 能够反对多正本的存储，而且数据可能主动迁徙实现正本均衡。咱们看到，Doris 中的数据是以 Tablet 的模式组织的，每一个表会拆分成多个 Tablet，每个 Tablet 是由数据分区跟数据分桶来确定的。一旦确定了 Tablet 之后，在 Doris 当中所有的数据都是基于 Tablet 来调度，咱们能够看到一个 tablet 能够扩散在多个 BE 上做多正本的存储，如果有 BE 节点宕机，或者是有新的 BE 节点退出时，零碎也会主动在后盾执行数据正本的平衡。
在查问的时候也会把查问负载平衡到所有的 BE 上，这就是 Doris 在数据正本存储上的整体架构，前面咱们做 Join 剖析的时候也会看到数据正本、包含数据是怎么样在当中调度的。
Doris Join 实现机制
Doris 反对两种物理算子，一类是 Hash Join，另一类是 Nest Loop Join。Hash Join：在右表上依据等值 Join 列建设哈希表，左表流式的利用哈希表进行 Join 计算，它的限度是只能实用于等值 Join。Nest Loop Join：通过两个 for 循环，很直观。而后它实用的场景就是不等值的 Join，例如：大于小于或者是需要求笛卡尔积的场景。它是一个通用的 Join 算子，然而性能体现差。
作为分布式的 MPP 数据库，在 Join 的过程中是须要进行数据的 Shuffle。数据须要进行拆分调度，能力保障最终的 Join 后果是正确的。举个简略的例子，假如关系 S 和 R 进行 Join，N 示意参加 Join 计算的节点的数量；T 则示意关系的 Tuple 数目。
Doris 反对 4 种数据 Shuffle 形式：
BroadCast Join
它要求把右表全量的数据都发送到左表上，即每一个参加 Join 的节点，它都领有右表全量的数据，也就是 T(R)。
它实用的场景是比拟通用的，同时可能反对 Hash Join 和 Nest loop Join，它的网络开销 N * T(R)。
Shuffle Join
当进行 Hash Join 时候，能够通过 Join 列计算对应的 Hash 值，并进行 Hash 分桶。
它的网络开销则是：T（R）+ T（N），但它只能反对 Hash Join，因为它是依据 Join 的条件也去做计算分桶的。
Bucket Shuffle Join
Doris 的表数据自身是通过 Hash 计算分桶的，所以就能够利用表自身的分桶列的性质来进行 Join 数据的 Shuffle。如果两张表须要做 Join，并且 Join 列是左表的分桶列，那么左表的数据其实能够不必去挪动右表通过左表的数据分桶发送数据就能够实现 Join 的计算。
它的网络开销则是：T（R）相当于只 Shuffle 右表的数据就能够了。
Colocation
它与 Bucket Shuffle Join 类似，相当于在数据导入的时候，依据预设的 Join 列的场景曾经做好了数据的 Shuffle。那么理论查问的时候就能够间接进行 Join 计算而不须要思考数据的 Shuffle 问题了。
上面这张图是 BroadCast Join，左表的数据是没有挪动的，右表每一个 BE 节点扫描的数据都发送到对应的 Join 节点上，每个 Join 的计算节点上都有右表全量的数据。
第二种就是 Shuffle Join，每个数据扫描节点将数据扫出来之后进行 Partition 分区，而后依据 Partition 分区的后果别离把左右表的数据发送到对应的 Join 计算节点上。上面这张图是 BroadCast Join，左表的数据是没有挪动的，右表每一个 BE 节点扫描的数据都发送到对应的 Join 节点上，每个 Join 的计算节点上都有右表全量的数据。
第二种就是 Shuffle Join，每个数据扫描节点将数据扫出来之后进行 Partition 分区，而后依据 Partition 分区的后果别离把左右表的数据发送到对应的 Join 计算节点上。
第三张图是 Bucket Shuffle Join，右表数据扫描进去之后进行数据分区的 Hash 计算，依据左表自身的数据分布发送到对应的 Join 计算节点上。
最初就是 CoLocate Join。它其实没有真正的数据 Shuffle，数据扫描之后进行 Join 计算就 OK 了。
下面这 4 种形式灵便度是从高到低的，它对这个数据分布的要求是越来越严格，但 Join 计算的性能也是越来越好的。
接下来就要分享的是 Doris 近期退出的一个新个性—— Runtime Filter 的实现逻辑。
Doris 在进行 Hash Join 计算时会在右表构建一个哈希表，左表流式的通过右表的哈希表从而得出 Join 后果。而 RuntimeFilter 就是充分利用了右表的 Hash 表，在右表生成哈希表的时，同时生成一个基于哈希表数据的一个过滤条件，而后下推到左表的数据扫描节点。通过这样的形式，Doris 能够在运行时进行数据过滤。
如果左表是一张大表，右表是一张小表，那么利用左表生成的过滤条件就能够把绝大多数在 Join 层要过滤的数据在数据读取时就提前过滤，这样就能大幅度的晋升 Join 查问的性能。
以后 Doris 反对三种类型 RuntimeFilter
一种是 IN—— IN，很好了解，将一个 hashset 下推到数据扫描节点。
第二种就是 BloomFilter，就是利用哈希表的数据结构一个 BloomFilter，而后把这个 BloomFilter 下推到查问数据的扫描节点。
最初一种就是 MinMax，就是个 Range 范畴，通过右表数据确定 Range 范畴之后，下推给数据扫描节点。
Runtime Filter 实用的场景有两个要求：
第一个要求就是右表大左表小，因为构建 Runtime Filter 是须要承当计算成本的，包含一些内存的开销。
第二个要求就是左右表 Join 进去的后果很少，阐明这个 Join 能够过滤掉左表的绝大部分数据。
当合乎下面两个条件的状况下，开启 Runtime Filter 就能播种比拟好的成果。
当 Join 列为左表的 Key 列时，RuntimeFilter 会下推到存储引擎。Doris 自身反对提早物化，提早物化简略来说是这样的：如果须要扫描 ABC 三列，在 A 列上有一个过滤条件: A 等于 2，要扫描 100 行的话，能够先把 A 列的 100 行扫描进去，再通过 A = 2 这个过滤条件过滤。之后通过过滤实现后的后果，再去读取 BC 列，这样就能极大的升高数据的读取 IO。所以说 Runtime Filter 如果在 Key 列上生成，同时利用 Doris 自身的提早物化来进一步晋升查问的性能。
上面简略比照一下三种不同类型的 Runtime Filter。
IN 的长处就是成果过滤成果显著，且疾速。它的毛病首先第一个它只实用于 BroadCast，第二，它右表超过肯定数据量的时候就生效了，以后 Doris 目前配置的是 1024，即右表如果大于 1024，IN 的 Runtime Filter 就间接生效了。
MinMax 的长处是开销比拟小。它的毛病就是对数值列还有比拟好的成果，但对于非数值列，基本上就没什么成果。
Bloom Filter 的特点就是通用，实用于各种类型、成果也比拟好。毛病就是它的配置比较复杂并且计算较高。
最初就是 Doris Join 的一个重要机制—— Join Reorder，在进行 Join 调优的时候会经常用到它。
数据库一旦波及到多表 Join，Join 的程序对整个 Join 查问的性能是影响很大的。假如有三张表 Join，参考上面这张图，右边是 a 表跟 b 张表先做 Join，两头后果的有 2000 行，而后与 c 表再进行 Join 计算。
接下来看右图，把 Join 的程序调整了一下。把 a 表先与 c 表 Join，生成的两头后果只有 100，而后最终再与 b 表 Join 计算。最终的 Join 后果是一样的，然而它生成的两头后果有 20 倍的差距，这就会产生一个很大的性能 Diff 了。
Doris 目前反对基于规定的 Join Reorder 算法。它的逻辑是：
让大表、跟小表尽量做 Join，它生成的两头后果是尽可能小的。
把有条件的 Join 表往前放，也就是说尽量让有条件的 Join 表进行过滤
Hash Join 的优先级高于 Nest Loop Join，因为 Hash join 自身是比 Nest Loop Join 快很多的。
Doris Join 调优实际
接下来就进入第三局部，Doris Join 的调优实际。第二局部分享了 Join 的机制之后，第三局部就须要利用 Doris 自身的一些 Join 个性，包含 Doris 提供的机制来做 Join 调优。上面是 Doris Join 调优的方法论：利用 Doris 自身提供的 Profile，去定位查问的瓶颈。Profile 会记录 Doris 整个查问当中各种信息，这是进行性能调优的一手材料。理解 Doris 的 Join 机制，这也是第二局部跟大家分享的内容。知其然知其所以然、理解它的机制，能力剖析它为什么比较慢。利用 Session 变量去扭转 Join 的一些行为，从而实现 Join 的调优。查看 Query Plan 去剖析这个调优是否失效。下面的 4 步基本上实现了一个规范的 Join 调优流程，接着就是理论去查问验证它，看看成果到底怎么样。如果后面 4 种形式串联起来之后，还是不见效。这时候可能就须要去做 Join 语句的改写，或者是数据分布的调整、须要从新去 Recheck 整个数据分布是否正当，包含查问 Join 语句，可能须要做一些手动的调整。当然这种形式是心智老本是比拟高的，也就是说要在尝试后面形式不见效的状况下，才须要去做进一步的剖析。
接下来通过展现几个理论的 Case，来分享一下 Join 的剖析调优流程。
看上面图上的 Profile，一个四张表 Join 的查问，通过 Profile 的时候发现第二个 Join 耗时很高，耗时 14 秒。进一步剖析 Profile 之后，发现 BuildRows，就是右表的数据量是大略 2500 万。而 ProbeRows（ProbeRows 是左表的数据量）只有 1 万多。这种场景下右表是远远大于左表，这显然是个不合理的状况。这显然阐明 Join 的程序呈现了一些问题。这时候尝试扭转 Session 变量，开启 Join Reorder。
set enable_cost_based_join_reorder = true
这次耗时从 14 秒降到了 4 秒，性能晋升了 3 倍多。此时再 Check Profile 的时候，左右表的程序曾经调整正确，即右表是大表，左表是小表。基于小表去构建哈希表，开销是很小的，这就是典型的一个利用 Join Reorder 去晋升 Join 性能的一个场景。
第二个 Case，存在一个慢查问，查看 Profile 之后，整个 Join 节点耗时大略 44 秒。它的右表有 1000 万，左表有 6000 万，最终返回的后果也只有 6000 万。
这里能够大抵的估算出过滤率是很高的，那为什么 Runtime Filter 没有失效呢？通过 Query Plan 去查看它，发现它只开启了 IN 的 Runtime Filter。
后面介绍了，当右表超过 1024 行的话，IN 是不失效的，所以基本起不到什么过滤的成果，所以尝试调整 RuntimeFilter 的类型。
这里改为了 BloomFilter，左表的 6000 万条数据过滤了 5900 万条。基本上 99% 的数据都被过滤掉了，这个成果是很显著的。查问也从原来的 44 秒降到了 13 秒，性能晋升了大略也是三倍多。
上面是一个比拟极其的 Case，通过一些环境变量调优也没有方法解决，因为它波及到 SQL Rewrite，所以这里列出来了原始的 SQL。
这个 Join 查问是很简略的，单纯的一个左右表的 Join。当然它下面有一些过滤条件，关上 Profile 的时候，发现整个查问 Hash Join 执行了三分多钟，它是一个 BroadCast 的 Join，它的右表有 2 亿条，左表只 70 万。在这种状况下抉择了 Broadcast Join 是不合理的，这相当于要把 2 亿条做一个 Hash Table，而后用 70 万条遍历两亿条的 Hash Table，这显然是不合理的。
为什么会产生不合理的 Join 程序呢？其实这个左表是一个 10 亿条级别的大表，它下面加了两个过滤条件，加完这两个过滤条件之后，10 亿条的数据就剩 70 万条了。但 Doris 目前没有一个好的统计信息收集的框架，所以它不晓得这个过滤条件的过滤率到底怎么样。所以这个 Join 程序安顿的时候，就抉择了谬误的 Join 的左右表程序，导致它的性能是极其低下的。
下图是改写实现之后的一个 SQL 语句，在 Join 前面增加了一个 Join Hint，在 Join 前面加一个方括号，而后把须要的 Join 形式写入。这里抉择了 Shuffle Join，能够看到左边它理论查问打算外面看到这个数据的确是做了 Partition，原先 3 分钟的耗时通过这样的改写完之后只剩下 7 秒，性能晋升显著。
接下来就依据明天分享的内容做一个最佳实际准则总结。次要分为 4 点：
第一点：在做 Join 的时候，要尽量抉择同类型或者简略类型的列，同类型的话就缩小它的数据 Cast，简略类型自身 Join 计算就很快。
第二点：尽量抉择 Key 列进行 Join，起因后面在 Runtime Filter 的时候也介绍了，Key 列在提早物化上能起到一个比拟好的成果。
第三点：大表之间的 Join，尽量让它 Co-location，因为大表之间的网络开销是很大的，如果须要去做 Shuffle 的话，代价是很高的。
第四点：正当的应用 Runtime Filter，它在 Join 过滤率高的场景下成果是十分显著的。然而它并不是万灵药，而是有肯定副作用的，所以须要依据具体的 SQL 的粒度做开关。
最初：要波及到多表 Join 的时候，须要去判断 Join 的合理性。尽量保障左表为大表，右表为小表，而后 Hash Join 会优于 Nest Loop Join。必要的时能够通过 SQL Rewrite，利用 Hint 去调整 Join 的程序。
后续布局
最初和大家介绍一件比拟激动人心的事。明天分享了很多 Join 调优实际的一些方法论，但很多 Join 须要去调优的起因，是因为 Doris 自动化调优做得不够好。接下来社区的重点工作方向之一是更智能的优化器，能够大幅缩小大家去做手动调优的心智老本。那么更好的智能优化器依赖于什么？是更全面的统计信息收集。所以社区在 Q3 会去做一个更粗疏的统计信息的采集工作，而后来进一步晋升查问优化器的自动化水平。心愿在公布下一个版本的时候，明天跟大家分享的这些内容大家都能够遗记它，这是一个咱们心愿 Doris 将来可能做到的事件。最初，在百度智能云上，百度提供了基于 Apache Doris 疾速迭代的企业级托管版本，感兴趣的小伙伴能够来尝试一下。感激大家的凝听，谢谢大家。