从 Clickhouse 到 Snowflake： MPP 查询层
从 Clickhouse 到 Snowflake： MPP 查询层
导语 | 伴随着Snowflake的成功，重新激活了数据分析市场，大大小小的创业公司不断创立，各种OLAP的开源产品层出不穷。其中，ClickHouse凭借优秀的性能在用户行为分析、ABTest、在线报表等多个领域大放异彩，但其在功能特性、易用性等方面都还有较多不足。同时，在OLTP、对象存储、Elasticsearch、MongoDB等系统中累积了大量数据和分析需求，不能较好的得到满足。因此，我们希望以Clickhouse为基础，借鉴Snowflake的设计思路，打造一款高性能的云原生OLAP数仓，为用户提供多数据源、多场景下的一站式数据分析平台。
MPP查询层核心特性
概述
- 功能强大，支持复杂的多表Join与聚合
- 内存零拷贝、全链路向量化的MPP实现
- 兼容SQL 标准 与 MySQL连接协议
- 持续兼容开源生态
背景
进入2021年，伴随着Snowflake的成功，大大小小的创业公司不断创立，各种OLAP的开源产品层出不穷，Clickhouse凭借优秀的性能在这其中脱颖而出，内部各种极致的优化，也被津津乐道，主要包括：
- 向量化思想，业界虽然很早就有向量化的理论，并且在各大公司的产品介绍中LLVM、向量化、SIMD这些光鲜的名词也屡见不鲜，但是Clickhouse 第一次把向量化这项技术在一个系统中做到极致，并被广泛使用，通过一个工程实践证明了向量化在OLAP 系统内靠谱、可行。
- 高质量的工程实现，数据库是一个系统工程，再好的理论也需要优秀的工程实现才能交付优秀的性能。Clickhouse把工程实现做到了极致，比如大量地使用模板技术来减少分支预测；Processor执行框架实现各个算子的异步执行，同时兼顾CPU Cache 的命中率。相信每个看过Clickhouse源码的人都会感叹“原来还可以这么用！”。此外，Clickhouse的编译依赖做的也非常棒，它把所有的依赖都以源码的形式引入到项目中从头编译，不需要用户下载任何其他第三方依赖，编译完之后是一个完整的、没有任何依赖的二进制库。
毫无疑问Clickhouse是一款追求性能极致的产品，但是在使用过程中我们发现它在功能和易用性上离通用的数仓（如Vertica，Greenplum等）还有一些差距，主要包括：
- 功能不足，多表Join支持差，用户一般需要使用大宽表；复杂的聚合容易OOM；缺少查询优化器的支持，用户需要手动调优；
- 兼容性不好，对SQL标准兼容弱，缺少一些常见的SQL 语法支持，比如没有SQL 相关子查询，这样很多现有工具不能直接使用，对大量SQL用户群也不友好。比如业务原来基于MySQL做BI 报表，如果想迁移到Clickhouse上，语法得改写，数据得重新建模。
- 易用性差，查询分为本地表查询和分布式表查询，比如在Colocate Join下用户就需要使用本地表，不易用。
为了打造一个媲美Snowflake的云原生数仓，为Clickhouse增加一个功能强大的的分布式查询层是我们必须要迈过的一道坎。
壹
云原生Clickhouse MPP查询层架构设计
增强Clickhouse的分布式查询能力，主要考虑过以下两种方案：
- 方案一，改进现有的查询层，在现在查询层的基础上，增加更多的SQL 语法支持来兼容SQL标准，嵌入实现一个查询优化器，完善现有的数据Shuffle逻辑等。这种方式可以复用Clickhouse当下优秀的计算能力，但是实现上想在不侵入Clickhouse源码的前提下改进扩充非常难，比如Clickhouse纯手工打造的SQL 解析器，想增加一条SQL 就需要改动很多模块。
- 方案二，实现一个全新的查询层，把Clickhouse完全当做单机引擎来用，查询层独立于Clickhouse，这样分布式查询层可以单独发展，不至于跟Clickhouse社区割裂。
与Clickhouse社区协同发展是保持产品生命力的重要方式，所以我们选择了方案二，架构如下图所示：
（查询层架构图）
- Master 节点，这个跟存算分离架构中的Master节点是一体的，由于在存算分离中，所有DDL 语句的执行都是通过Master节点来调度执行的，所以Master节点在执行DDL 任务的过程中通过解析DDL SQL 建立了全局一致的Catalog；Master节点内部还包括一个SQL 优化器，来生成高效的物理查询计划；
- CK 节点， 在Processor的执行框架之上，增加了一个MPP 模块，用来实现一套功能强大的分布式MPP 查询层；
- 共享存储，实际的数据文件和CommitLog文件都存放在对象存储中，对象存储目前仅支持腾讯云对象存储COS，这部分在存算分离相关文章中有介绍。
在该架构下，查询的执行流程如下图所示：
（查询执行流程图）
- 用户可以随意连接一个Clickhouse节点，发送SQL语句；当前这个Clickhouse节点作为本次查询的Initiator，把查询转发给Master；
- Master节点根据Catalog中的Schema做查询SQL的解析，根据数据分布来生成物理查询计划；
- Initiator拿到物理查询计划之后，分发给对应的Clickhouse 节点执行；
- 各个Clickhouse 节点通过 MPP 执行模块完成分布式查询的执行，包括调用Clickhouse的Processor 扫描数据， 跟其他节点进行数据交换，执行各种算子（例如Join，Agg等）；
- 最终的结果返回给Initiator，Initiator把结果根据不同的协议进行格式化，返回给客户端；
整个查询的执行过程中，数据流不经过Master节点，降低Master节点的压力；Master单节点可以支撑万级QPS的查询解析请求，由于Master节点多副本，所以可以通过集群的方式进行线性扩展。
贰
云原生ClickhouseMPP查询层核心特性
功能强大的MPP计算引擎
Clickhouse的执行框架是一个简单的两阶段执行框架（Scatter Gather 模型），Scatter 任务可以由多个节点来完成，Gather 任务只有一个节点来完成，所以它不能执行复杂的查询，比较适合大宽表。而业界典型高性能查询引擎使用的MPP计算框架是一个多阶段的执行框架，一条复杂的SQL 语句被拆解为多个计算算子，每个计算算子可以分布到多个计算节点上并行完成，计算节点之间通过RPC 完成数据交换，并以Pipeline的方式高效执行。
（执行框架对比图）
下面可以通过大宽表上的一个简单查询来看一下两者之间的区别，SQL 语句如下：
SELECT age，count(distinct uid) from user_info group by age
Clickhouse的执行流程如下：
- Scatter阶段 ：Initiator 节点向各个 Shard 发送查询，要求其返回执行到 WithMergeableState 的结果，该阶段包含聚合逻辑的前半部分，相同 Key 已在哈希表中聚合，Value 仍保持可聚合的中间状态。该中间状态可以被序列化（Clickhouse 特色的中间结果-），并返回给 Initiator。
- Gather阶段 ：Initiator 继续执行聚合逻辑的后半部分，将相同 Key 的所有中间状态合并，并转换成最终的值。合并过程只能在单点完成。
MPP 框架的执行流程如下图所示：
（MPP框架执行流程图）
这个查询语句被规划为3个阶段， 扫描数据， 聚合计算，返回结果；每个阶段又会被拆分为多个子任务，例如这个查询就被拆分为7个任务。
- 阶段一：3个节点同时执行Scan 任务，每个任务执行完之后，把数据根据Age字段Hash分桶，分别发送给3个不同的Agg任务；
- 阶段二：3个Agg 任务根据收到的数据，按照Age来做Group by计算，把结果发送给Sink任务；
- 阶段三：Sink 任务收到的数据已经是聚合好的，所以可以直接对数据进行简单的Merge，然后返回给客户端。
与Scatter-Gather模型相比，上述聚合计算被分配到多个节点上并行执行了，不仅仅可以加快速度，还可以降低内存使用，避免内存不足。在具备通用的MPP 执行框架之后，已经可以跑通Join等大多数复杂查询，后续通过查询优化器合理的查询规划，可以进一步提升复杂查询的性能，基于代价的查询优化器（CBO）正在研发中，预计下一个版本发布。
内存零拷贝、全链路向量化的MPP实现
业界有很多MPP查询引擎的实现，比如Impala，Presto，Spark等，我们看到很多公司也在尝试将这些查询引擎对接Clickhouse，从而让Clickhouse具备MPP 执行的能力，但是从调研分析看，这种方式有以下缺陷：
- 数据传输开销大，Clickhouse作为存储层与查询层在两个服务进程中（非混部场景中，在两台机器上），数据的传输需要序列化和反序列化，跨网络或者单机多进程之间传输，开销比较大；
- Clickhouse 相比其他OLAP 系统很大的优势在于它向量化的思想以及高质量的工程实现，当查询层交由别的系统来实现之后，Clickhouse就只剩下单机的扫描能力，强大的查询能力就发挥不出来了。
而Clickhouse 最大的优势就是快，这种整合方式会让Clickhouse丧失这个优势，产品竞争力就会下降，而且交付给客户的是一个多个组件构成的“全家桶”，使用起来也复杂。所以我们抛弃了这种方式，选择在ClickHouse同进程内、Processor执行框架之上实现MPP 查询层，如右下图所示：
（实现MPP查询层）
方案的整体思路及优势如下：
- MPP 计算层跟Clickhouse在同一个进程内，不需要序列化传输数据；
- MPP 计算层也是用Block作为内存数据格式，与存储层之间的数据交换不需要内存拷贝，进一步减少开销。
- MPP 计算层在Block的内存结构之上，复用Clickhouse的向量化计算的算子，达到跟Clickhouse同样的性能；
- MPP 计算层把简单的函数表达式计算、过滤等算子全部下推给Clickhouse完成，未来单表上的聚合也会下推给Clickhouse，充分利用Clickhouse的索引、统计信息、并行聚合等优势能力；
这种设计在兼容开源、保持简洁的同时，尽可能做到零序列化、零拷贝，并充分复用ClickHouse的向量化算子等能力，保持ClickHouse的最大优势 “快”。
兼容SQL 标准 与 MySQL连接协议
充分利用当前的SQL 与 MySQL生态，应用程序无需修改即可切换到Clickhouse服务上，享受Clickhouse带来的极速的分析能力。目前我们已经能够在不需要改造大宽表模型下，完全跑通TPC-H的所有测试语句，TPC-DS标准也支持了90%以上。例如TPC-H Q21这种复杂的多表Join和子查询场景：
能够支持常见的BI 工具，例如业界排名第一的Tableau，用户可以选择MySQL连接，直接当做MySQL来使用即可。如下图所示：
持续兼容开源生态
在实现MPP查询引擎时，我们仍然遵循着不侵入Clickhouse源码的原则，把Clickhouse当做一个单机的库，如下图所示：
- 在底层，我们用存算分离替换了Clickhouse的本地存储；
- 在上层，用MPP 查询层替换了Clickhouse当前的查询框架；
- 在周边，我们利用Clickhouse的SQL 命令实现了全新的分布式DDL 框架；
- 屏蔽Local表的导入功能正在研发中。
这种架构使得后续的版本升级更加方便，能够随时合并Clickhouse社区的最新功能。
兼容开源生态的另外一个方面是新系统允许用户在MPP 查询引擎和现有的查询引擎之间切换，用户可以通过一个Setting 完成，如下所示：
SET use_mpp_engine = true
用户可以在新场景里使用MPP 查询引擎，逐步的把Clickhouse目前的查询语法废弃，平滑升级到新的查询引擎，未来我们也会在MPP查询引擎中兼容Clickhouse的SQL 语法标准，让用户的迁移更便利。
未来工作
- 本地Cache优化，存算分离架构中本地Cache实现的好坏对性能有决定性影响，这是我们近期要重点攻克的地方。这部分完成后，我们会发布正式的性能测试报告。
- CBO 查询优化器，这是执行复杂查询必备的一个组件，目前我们正在开发中，预计明年上半年上线。
- 多数据源支持，在OLTP、对象存储、Elasticsearch、MongoDB等系统中，存在大量数据需要进行深入分析。借助Clickhouse当前的多数据源能力，支持对多数据源、结构化/半结构化数据的统一分析。
- 简单完全的分布式化，消除节点、Shard等用户概念，给用户暴露一个高度抽象、类单机的分布式系统，屏蔽底层的实现细节，降低用户的使用负担或顾虑。