腾讯天穹SuperSQL联邦融合计算引擎揭秘

如果无法正常显示，请先停止浏览器的去广告插件。

1. 「解耦」方能「专注」—— 腾讯天穹SUPERSQL跨引擎计算揭秘陈奕安

2. 目录 CONTENT 01 02 03 04 背景目标业界竞品系统架构元数据管理 05 06 07 08 跨源SQL算子下推分布式计算引擎跨DC查询优化性能评测

3. 01 背景目标

4. SuperSQL项目背景：问题与挑战异构数据源不同版本数据源系统日志用户行为 1.x 2.x 数据不互通，迁移困难（海量、业务、安全等）用户基本信息现状: 应用而非平台层级的SQL拆分，人工数据搬迁与查询调优数据源位于不同地域的多个 DC/ 集群无统一逻辑视图带宽受限，端口 /IP 屏蔽

5. SuperSQL落地成果：支持跨业务平台（一）支持跨业务平台的数据融合，提升业务效率广告某业务场景(跨业务平台的数据分析) l 业务需求：联合查询/分析存储在私有集群的数据和中心化数据仓库中的业务数据，分析在中心化数据仓库 l 业务痛点： ü 数据导入中心化数据仓库费时费力 ü 数据一致性无法保证 ü 缺少统一的数据查询/分析入口 l l l l SuperSQL效果无需数据搬迁：节省开发人力5-10人，节省导入时间1~2h 查询分析效率：性能总体提升3倍以上统一数据查询分析入口：SuperSQL管理所有的数据源整体一致性保证：避免了ETL操作，一份数据，无需跨多个系统来验证备份数据的一致性 SQL分析 SuperSQL 数据量约60TB 单表最大100亿条千万级别数据数据导入某业务用户私有数据 HIVE 数据导入数据源数据源 Hive 某业务用户私有数据数据源

6. SuperSQL落地成果：支持跨业务平台（二）支持跨业务平台的数据融合，提升业务效率某业务场景(跨业务平台的数据分析) l 业务需求：联合查询/分析存储在私有集群的数据和历史数据仓库中的业务数据，分析在业务平台 l 业务痛点： ü 数据出库到业务平台流程复杂 ü 数据任务脚本（Spark jar）维护困难数据出库 HDFS 用户在历史数仓的数据 SuperSQL效果 l 流程简单：数据到业务平台只需 SuperSQL，数据无需中间落地，节省导入时间约1h l 维护简单：数据任务由SQL完成，逻辑清晰 SuperSQL 数据导入业务平台用户在历史数仓的数据业务平台

7. 02 业界竞品

8. 业界竞品分析：开源软件 SparkSQL l l l 简介：DataFrame/DataSet编程抽象，分布式SQL查询优势：原生HDFS访问优化，DataSource API对接异构数据源，Catalyst SQL Parser/RBO，Stage失败重试不足：通用DAG而非专用SQL引擎， DataSource V1 仅支持简单Project/Filter下推数据源， DataSource V2尚待完善，CBO雏形且发展不明朗 Presto l l l 简介：基于内存的并行计算，Facebook推出的分布式SQL交互式查询引擎，多个节点管道式执行优势：MPP模型，仅计算部分数据放置内存，多Connector支持异构源不足：不适合多个大表长时间的Join操作，全内存操作容易OOM，容错机制较差 Drill l l l 简介：Google Dremel系统开源实现， MapR出品，低延迟交互式SQL分析优势：基于Calcite自研实现的分布式SQL 执行引擎，MPP模型，P2P架构不足：多源对接功能基本源于社区Calcite 无扩展，过于聚焦HDFS加速，稳定性和社区生态无法匹及Spark/Presto

9. 业界竞品分析：开源扩展 Pollux (未开源) l l l 简介：跨DC多数据源的协同计算，基于开源SparkSQL深度定制扩展优势：复杂算子下推，跨DC CBO，DSA传输加速，Spark任务日志不足：可扩展性不太好，与Spark代码强耦合(侵入式修改)，无Join下推支持，CBO依赖于Spark社区尚未成熟的方案 Quicksql (开源) l l l 简介：基于Calcite扩展实现，具备多数据源的接入能力，解析转化用户提交的查询语句，最终提交到Jdbc/Spark/Flink执行优势：整体的架构方案做到了轻量级，提供单 DC CBO与外部计算引擎对接不足：核心功能基于开源Calcite，未解决社区版本的一系列问题，例如多Jdbc源SQL仅单源能下推、复杂查询无法生成可执行计划或CBO 耗时过长等；开源版本状态几乎无法商用，元数据机制、Spark/Flink对接代码均仅达到Demo 级别；不支持跨DC CBO与查询执行

10. 业界竞品分析：私有系统 F1-Query (未开源) l l l 简介：Google内部异构数据源查询引擎，支持对各种不同文件格式和存储系统 (Bigtable, Spanner, Google Spreadsheets) 中的数据进行跨DC的联合查询优势：各模块(如Metadata、UDF统一管理) 的抽象设计具有较为通用的参考价值不足：仅用于Google内部OLTP场景，私有系统参考文献较少 Data Lake Analytics (未开源) l l l 简介：阿里自研的数据湖产品，为用户提供跨源的联合分析能力，内部自研组件 + Presto扩展优势：支持多路数据源接入，提供多样化、异构数据源关联分析能力，融合MPP和DAG技术，具备横向分析扩展、向量化执行、算子流水线优化等能力不足：无法灵活地管理数据源间的层级关系，资源管理、任务调度执行等重复“造轮子”，技术架构针对特定场景的定制化程度高，通用适配性一般

11. 业界竞品分析：华为河图（HetuEngine）数据虚拟化引擎：像使用 “数据库”一样使用“大数据”，Pollux衍生扩展 HetuEngine引擎可对底层各数据中心的RDBMS、HDFS、 ElasticSearch、Redis、In- Memory DB、Kafka等进行统一的数据管理和治理 l 核心能力 n n n n “一个目录”：数据全域共享、打破数据孤岛、支持1000+数据源 “一个接口”：统一SQL接口、降低开发成本，支持5000+节点 “一份数据”让数据零搬迁、提升时效性、基于CarbonData “统一安全”完整保护数据、保障安全开放、细粒度动态授权 l 当前为私有系统，公开资料甚少，计划2020年6月内核开源，尚无性能数据 n n n “开发者可以基于开源代码进行定制，包括数据源扩展、SQL执行策略等” “确保合作伙伴们能够免厂商锁定” “复用现有的生态、工具和技能，提升开发效率2到10倍” 北向应用接口 + 南向数据接口 + 内核引擎

12. 业界竞品分析：Contiamo Workbench l 德国“Start-up”大数据公司，成立于2013年，CTO是Calcite Committer l Calcite扩展，使用Spark作为外部计算引擎，技术方案基本与SuperSQL一致 l UI：类似IDEX，更为丰富的可视化功能组件，提升用户体验表列统计信息数据源列表执行计划工作流支持数据源类型（混合云） SuperSQL 产品化指引

13. SuperSQL vs. 商业竞品：更丰富完善的功能维度 DLA ( 阿里 ) NoSQL：TableStore、 OSS、MongoDB、Redis 数据源 RDBMS：MySQL、 PostgreSQL、SQL Server、Oracle MySQL 5.7绝大部分查询语法 Hetu ( 华为 ) Pollux： Hive、 MPPDB、Oracle、 MySQL RDBMS：Oracle、 MySQL、PG、SQL Server、Hana、 DB2 SuperSQL RDBMS：MySQL、 PostgreSQL、Oracle、 ClickHouse … Cloud：S3、Azure NoSQL：Hive、Phoenix、 ES、Kylin、Hermes … Spark-SQL文法 SQL-2003标准 SQL-2003标准内部映射 Spark Catalog 未知独立模块，Hive MetaStore持久化查询优化 ① 支持OSS Select单文件的计算下推优化 ② 针对Table Store SDK 进行列式压缩调用返回优化 ① 计算下推优化： Aggregate/OrderBy/Li mit/复杂表达式下推 ② 跨DC查询的CBO优化基于规则的优化器 (RBO)，支持各类算子下推 ① 基于规则的优化器(RBO)，支持各类算子下推 ② 多阶段Planner规则切分 ③ 单DC内代价优化器(CBO) ④ 跨DC的代价优化器(CBO) 计算引擎自研（/Presto） Spark(/Presto…) Spark Hive/Spark/Livy/… 智能选择支持SQL 元数据 Hetu：ES、Redis、 Kafka Contiamo 技术断裂点 / 趋势：平台、成品、实战跨源：类型多样性，语法兼容性，系统稳定性跨 DC ： CBO 因子，子查询分布式协同执行跨引擎：耦合度，性能，升级 / 替换灵活性，负载 / 场景适应性

14. 03 系统架构

15. SuperSQL内核：Apache Calcite数据管理框架 l 工业级的SQL Parser与Validator l 关系代数变换：SQL -> SqlNode -> RelNode -> SQL l RBO/CBO框架：RelOptRule + HepPlanner & VolcanoPlanner l 元数据接口定义：Schema/Table/… l 多类数据源连接器 (Adapter) l JDBC Client/Server (Avatica) l 单机单线程Enumerable执行引擎 (Linq4J) Calcite 局限性：框架而非平台，原料而非成品，注重功能示例而非实战使用 SuperSQL: Calcite ++

16. SuperSQL整体架构：漂移计算 Query: (SELECT t1.a, count(t1.b) as cnt FROM dc1.tbase.db1.tbl AS t1 WHERE t1.c > 100 GROUP BY t1.a ORDER BY t1.a LIMIT 10) UNION ALL (SELECT t2.a, count(t2.b) as cnt FROM dc2.hive.db2.tbl t2 GROUP BY t2.a LIMIT 20) L0 Calcite Avatica L1 Calcite 应用 SQL JDBC Driver l SuperSQL：联邦政府 l DC(数据中心)：各个“自治州”政府 l 用户SQL：涉及多个自治州的任务，统一提交联邦政府处理，并等待接收结果 Apache Ranger l 认证鉴权服务 Result 子SQL下推：将单个自治州能处理的子任务下发，联邦政府只需要结果 l SuperSQL Server 汇总计算：由挑选的“中心州”代理，各个自治州子任务处理完后，中心州负责结果的整理与汇总，最后 DC内/跨DC CBO 统一元数据存储下推子SQL生成 Hive Metastore 选择最佳DC执行返回给联邦政府 l DC内CBO：各自治州内事务如何处理最有效率？ l 跨DC CBO：汇总计算，放到哪个中心州比较合适？中心州是否挑选多个协同处理更为合适？漂移计算计算引擎 L2 角色类比联接合并逻辑视图/临时表 Spark DataSource API / Hive Jdbc StorageHandler 计算引擎改写 SQL: (SELECT * FROM dc1_pg_db1_tbl_UUID) UNION ALL (SELECT * FROM dc2_hive_db2_tbl_UUID) 漂移计算 L 3 数据源 Hive HBase DC 2 DC 1 …… TBase Hive PG DC n MySQL ES Oracle 漂移计算 View 1: dc1_pg_db1_tbl_UUID SELECT t1.a, count(t1.b) as cnt FROM dc1.tbase.db1.tbl AS t1 WHERE t1.c > 100 GROUP BY t1.a ORDER BY t1.a LIMIT 10 View 2: dc2_hive_db2_tbl_UUID SELECT t2.a, count(t2.b) as cnt FROM dc2.hive.db2.tbl t2 GROUP BY t2.a LIMIT 20

17.

18.

19. 04 元数据管理元数据 = 数据源信息 + CBO 统计信息

20. 数据源元数据模型元数据模型设计 l 树结构：特殊的Trie，每个节点对应的子树形成了一个子命名空间 l 树节点： CatalogTreeNode主要内容有父节点parent、子节点列表children、节点路径path、名称name (path中最后一个元素)，以及Map类型的conf配置信息 select * from test.mytable select * from dc1.subdc3.pg.test.mytable 虚拟 root 节点 CatalogTreeNod e [“”] 数据源数据表 CatalogTreeNode [“dc1”] CatalogTreeNod e [ “mydb”] CatalogTreeNod e [ “subdc3”] CatalogTreeNode [ “mysql”] CatalogTreeNod e [ “pg”] CatalogTreeNode [ “test”] CatalogTreeNod e [ “test”] CatalogTreeNod e [ “dc2”] CatalogTreeNod e [“test”] CatalogTreeNod e [ “mydb”] CatalogTreeNod e [ “hive”] CatalogTreeNod e [ “mydb”] CatalogTreeNod e [ “pg”] CatalogTreeNod e [ “mydb”]

21. SuperSQL元数据管理I：层级数据源表征 l 通用性：解决传统的 DB/Table 模型面临的同库同表等问题 l 灵活性：数据源定义需要非常灵活，上下层级关系可任意定义，物理世界实体的逻辑抽象（主数据中心 -> 子数据中心 -> 机架 -> 集群 -> 数据库系统） l 轻量级：最大化地复用现有开源组件，避免引入额外维护开销

22. SuperSQL元数据管理II：CBO统计信息采集 l CBO需要估算SQL执行计划中每个算子的代价，包括CPU、IO、Network 等 l 一个物理计划的代价，就是其树型结构上所有节点（算子）的代价之和 l 在无数候选计划中，CBO挑选代价最小的计划，作为最优计划来执行 l 估算某个SQL算子（如Filter、Join、Sort）的代价，需要首先预估其输出和处理的行数/字节数，其基础为每个底层Scan算子的预估值，通过采集对应数据源表的统计信息（CBO Stats）来计算 l 数据源表的统计信息分为表级和列级两类自动生成最优的执行 n 表级：行数，字节数 n 列级：数值型 ü Histogram ü Selectivity CBO/RBO 字符串型用户只关心业务逻辑不懂 SQL 优化，不关心 SQL 不同写法社区Calcite有Statistics接口, 但无统计信息采集和使用的逻辑实现

23. 统计信息使用样例：Filter输出估算假设uniform distribution 若存在直方图，估算会更精确先估算输出rowCount，然后根据各列的平均长度计算sizeInBytes

24. 统计信息持久化：Hive MetaStore l 当前SuperSQL对接Hive Metastore方式问题：同一数据源的所有表的 Cbo Stats 会被同时读写，效率低下，造成 Hive MetaStore 性能压力 u 专用一个名为“supersql”的数据库 u 每个注册到SuperSQL 的数据源（DataSource），对应该数据库中的一张虚拟表 u DataSource属性（config）存入Hive Metastore的TABLE_PARAMS表中 l CBO Stats使用类似方式存入Hive Metastore u u u u srcTableName 是数据源中某张数据表的名称参数名：supersql.cbo.tableStats.srcTableName.segment.X 参数值：TableStat（含ColumnStats）对象序列化为字节数组后，base64编码因参数值有最大长度限制，分段（segment）存储（X = 1, 2, …）特殊参数 “supersql.cbo.tableStats.srcTableName.segment.-1” 存储segment总数 Supersql DataSource 对应的 Hive 虚拟表 ID Stats 存储格式样例 TBL_ID PARAM_KEY PARAM_VALUE 1 supersql.cbo.tableStats.table1.segment.0 XXXXXXXX… 1 supersql.cbo.tableStats.table1.segment.1 XXXXXXXX… 1 supersql.cbo.tableStats.table1.segment.2 XXXXXXXX… 1 supersql.cbo.tableStats.table2.segment.-1 2 2 supersql.cbo.tableStats.table3.segment.0 XXXXXXXX…

25. 统计信息采集：Analyze 命令 ANALYZE TABLE datasourceName.tableName COMPUTE STATISTICS [FOR COLUMNS columName1, columName2, …] [WITH SAMPLE SIZE sampleNum] [WITH SAMPLE TIMES sampleTimes] [WITH INCREMENT condition | RECOMPUTE] 手动执行，非系统自动触发，可配置定期脚本 l 扩展Calcite Parser实现，单一命令同时采集表级 + 列级统计信息 u u 表级Stats必采：简单、高效若不包含FOR COLUMNS子句，默认采集所有列的Stats；若包含则仅采集指定名称的列，至少需指定一列 l WITH SAMPLE SIZE指定采样的行数 u 默认值设置为 N = min(tableRowCount, 1000w) u 全表随机采样，如果数据源表包含分区且该信息可以通过Jdbc接口获取，则每个分区均随机采样1/N l WITH SAMPLE TIMES指定采样的次数，多次采样可以增加随机概率 u 默认值为1，多次采样的结果增量合并 u 累积采样效果，提升Stats精度，从而潜在提升CBO的精度 l 同一数据表先后执行多次Analyze命令，默认更新模式是RECOMPUTE u RECOMPUTE子句（可选）指定本次命令指定为全量覆盖，即删除不合并之前采集的Stats，适用于表数据变化较大或历史Stats采集明显不准确的情况 u WITH INCREMENT子句（可选）指定一个条件，如dp = ‘2019-08-01’ and cp > 100，命令执行时仅随机采样满足该条件的数据，且采样结果增量合并入当前Stats，适用于新增或删除表分区的情况

26. 列统计信息采样与合并逻辑列 Stats 采集函数合并逻辑（前后两次采样） ndv COUNT(DISTINCT col) 按采样行数比例求均值，ratio = N1 / (N 1 + N2），ndv = ndv1 * ratio + ndv2 * (1 - ratio) nullCount SUM(CASE WHEN col IS NULL THEN 1 ELSE 0 END) 按采样行数比例求均值 max MAX(col) MAX(max1, max2) min MIN(col) MIN(min1, min2) maxLen MAX(LENGTH(col)) MAX(maxLen1, maxLen2) avgLen AVG(LENGTH(col)) (avgLen1 * rowCount1 + avgLen2 * rowCount2) / (rowCount1 + rowCount2) selectivity SUM(CASE WHEN col THEN 1 ELSE 0 END) / COUNT(1) (selectivity1 * rowCount1 + selectivity2 * rowCount2) / (rowCount1 + rowCount2) DISTINCT col，SuperSQL端通过HyperLogLog （HLL） [4] 来计算和表示，每个直方图 bucket对应一个HLL数据结构 1. SuperSQL采用近等高直方图，即不强制要求所有bucket中的值数必须相等 2. 后续采样沿用前次采样的bucket边界，同边界的两个 bucket对应的HLL merge 3. 最大最小bucket的高度差，若超过系统定义阀值则需部分调整（合并前后bucket），若调整失败则全量重构直方图 histogram

27. 05 跨源 SQL 算子下推

28. SuperSQL：一站式高性能大数据SQL中间件 l 跨源：支持访问不同类型/版本的数据源 – Hive // SparkSQL / PG / TBase / MySQL / Oracle / H2 / HBase (Phoenix Query Server) / ElasticSearch (sql4es) / Kylin / ClickHouse / Druid / PrestoDB / PrestoSQL / SuperSQL l 跨引擎：支持外接多类分布式计算（SQL执行）引擎 – Hive（2.3.x）/ Spark（ Livy、Kyuubi）/ Flink / Presto l 跨DC：支持跨数据中心/集群的SQL优化与执行 – RBO/CBO 面向跨DC异构数据源的统一数据分析平台内外部需求产品化销售历史包袱现网业务支撑融合其它开源据工厂开源协同

29. 计算下推：支持常用复杂SQL算子扩展60+下推规则实现 • • • 网络开销大性能低：计算引擎通过JDBC单线程拉取大量数据，网络IO大且速度慢计算引擎负载过重：计算逻辑都在汇总的计算引擎完成，其所在集群负载压力大资源使用不合理：数据源的计算资源处于闲置状态，未能充分利用 SQL 算子类型包括count、sum、average、max、min等 Top N (Limit/Offset) 不同数据源语法不同，如Oracle为rownum < N 排序 (OrderBy) 去重可以减少数据量系统标量函数(Scalar Function) length、nvl、datetime、power等 CBO估算Join的结果是否会膨胀，权衡传输与计算开销；Join的重排序 Join 数据量爆炸数据源计算 select a1, a2 from t1 where a1 > 10 Sort Having Aggregate JDBC JDBC Project Project Filter Filter Scan Scan 数据源1 数据源2 下推到数据源执行，如tHive UDF/UDAF 数据量可控 Join Limit Order-By不下推对应Limit无法下推 Union/Intersect/Minus 数据源专有函数计算引擎说明聚合函数 (Aggregate) 解决方案 (算子下推) 数据源计算 select b1, b2 from t2 where b2 < 1 下推 SQL 算子类型可通过参数灵活配置数据源计算 select a1, a2 from t1 where a1 > 10 order by a3 limit 20 计算引擎 JDB C Limit Sort Project Filter Scan 数据源1 Join JDB C Having Aggregate Project Filter 数据源计算 select b1, b2, max(b3) Scan 数据源2 from t2 where b2 < 1 group by b1, b2 having sum(b4) > 100 29

30. Join下推示例原始计划树 Join (t1.a = t2.b) Join Project Filter t1 Scan 变换后计划树 (t2.b = t3.c) Join Project Filter Filter Scan Join (t1.a = t3.c) Project Project (t1.a = t2.b) Project Proje ct Filter Filter Scan Scan Filter Scan Scan t2 t1和t2来自不同数据源，无法下推这里假设Join的执行次序一定是从左到右。 t3 t1 t3 假设t1和t3来自同一数据源。 Join重排序过程中，还要考虑Join条件的自动传递，当前只考虑Equality条件。 t2

31. Aggregate下推Union All（1/2）仅考虑最常用的 count/sum/avg/max/min 聚合函数，不支持 count distinct t1 和 t2 来自不同 DC 的数据源 select count(*) from (select id from t1 union all select id from t2) t select sum(cnt) from (select count(id) as cnt from t1 union all select count(id) as cnt from t2) t select max(id) from (select id from t1 union all select id from t2) t select max(id) from (select max(id) as id from t1 union all select max(id) as id from t2) t select avg(num) from (select num from t1 union all select num from t2) t select sum(s)/sum(c) from (select sum(num) as s, count(num) as c from t1 union all select sum(num) as s, count(num) as c from t2) t 不能是 Union ，因为涉及去重时无法确保语义正确好处：能大量减少数据传输

32. 下推SQL优化：并发子查询切分动机：当下推某一数据源的SQL返回数据量很大时（如千万条记录，可通过CBO估算），单个JDBC连接获取结果耗时长适用场景：基于统计信息，发现数据源表有合适的分区（如Hive）或Unique索引（如TBase）当前局限：只支持简单 Select- Where 查询，不支持带 Agg./OrderBy/ Limit/Join 的查询（语义复杂，无法确保拆分子查询结果合并的正确性） select a, b, c from t1 where a > 10 or d < 20 select a, b, c from t1 where (a > 10 or d < 20) and parCol < 10 select a, b, c from t1 where (a > 10 or d < 20) and parCol >= 10 and parCol < 20 select a, b, c from t1 where (a > 10 or d < 20) and parCol >= 20 and parCol < 30 原理： 1. 根据分区、索引或其它列统计信息，为下推数据源表挑选一个 “最好的切分列”parCol 2. 基于parCol的最大最小值生成N个disjoint的分区条件（N可以固定也可以动态根据数据分布确定） 2. 将分区条件拼接到原始下推SQL的Where条件中，生成N个对应的子查询 3. 并发多线程执行N个子查询，合并ResultSet

33. 06 分布式计算引擎

34. SuperSQL分布式计算引擎需求 l 快速构建：不重复“造轮子”，尽量复用通用开源软件 l 轻量级解耦：不强依赖特定框架，不做或少做侵入性修改 l 场景适配：根据SQL特征，智能挑选主流执行引擎

35. 计算引擎处理流程：数据源下推子计划 ->临时表映射 SuperSQL CBO-> 通知计算引擎 (二次优化) -> 数据源执行 -> 计算引擎汇总 SuperSqlImplementor 计算引擎 select name, cnt EnumerableCalc from view1 join view2 EnumerableJoin on view1.a=view2.a 执行计划 SuperSqlImplementor PostgreSQL Dialect View1 View2 JdbcToEnumerableConverter JdbcToEnumerableConverter JdbcAggregate JdbcJoin JdbcTableSca n JdbcTableScan CREATE TEMPORARY VIEW view1 USING org.apache.spark.sql.jdbc OPTIONS (url ‘jdbc:postgresql://host1:port1/db1’, dbtable ‘(select a, count(*) as cnt from t1 group by a) t’, user “******", password “******"); 数据源 SuperSQL 外接执行引擎要求： 1. 提供 JDBC Client/Server 2. 能够通过 JDBC/HDFS/ 定制 Connector 连接到各类 SuperSQL 数据源 3. 支持动态增加 / 删除数据源或数据源表 SuperSqlImplementor JdbcTableSca n MySQL Dialect CREATE TEMPORARY VIEW view2 USING org.apache.spark.sql.jdbc OPTIONS (url "jdbc:mysql://host2:port2/db2", dbtable ‘(select a, name from t2 join t3 on t2.id = t3.id) t’, user “******", password “******");

36. 07 跨 DC 查询优化

37. SuperSQL CBO概览 SuperSQL CBO Calcite CBO 单阶段 VolcanoPlanner 单因子 (rowCount) 算子代价估算多阶段Planner (Volcano/Hep) 超时机制死锁检测 Stats采样/增量收集 (Analyze) 多因子（rowCount/sizeInBytes/ network bandwidth）算子代价估算下推并发子查跨DC最优计跨DC分布式Join 询切分算引擎

38. DC内CBO：多阶段Planner • 多阶段Planner CBO比单阶段约快5 倍超时机制：加入软硬超时阀值，禁用计算开销大的规则或强制终止，可选次优执行计划 • 多阶段Planner：根据变化目标，将规则集合拆分成多个串行阶段，每个阶段对应独立的规则子集 • 验证逻辑：检测相互冲突的规则，避免形成“死锁” 似的循环规则匹配 CBO 平均耗时 (ms) 多阶段 Planner 单阶段 … 0 2000 4000 6000 单阶段 VolcanoPlanner 多阶段Planner SubQuery Decorrelate Pre-Trim Logical Optimization Join Reordering Aggregate Pushdown Enumerable Optimization 单一队列，重复冗余匹配复杂 SQL CBO 卡住数十分钟，经常无法生成可执行计划 Plan Validation Post-Trim Final Calc

39. 跨DC CBO：数据源间网络带宽在创建 / 修改数据源的 SQL 命令中，通过 bandwidth 可选参数，指定该数据源到其它数据源之间的静态网络带宽 CREATE/ATLER DATASOURCE dc1.cluster1.hive1.db1 with (datasource=jdbc, catalog=tpcds_1g, …, bandwidth=`[dc2:1g, dc3.cluster2:100m]`) 100m bit/s dc2 dc1 dc1.cluster1 dc2.cluster1 dc1.cluster1.hive1 dc2.cluster1.tbase dc1.cluster1.hive1.db1 dc1.cluster1.tbase.db1 基于所有DataSource相关属性值，构建数据源带宽图 l 每个合法的数据源名前缀，均对应一个节点 l 所有节点到其它任何节点均有边，边的权重代表数据从节点A发送到节点B的单向网络带宽 l 如果A到B的带宽已指定，但B到A的带宽未指定，则认为带宽对称 l 如果A到B的带宽未指定，A到B的带宽定义为A的父节点到B的带宽 l 如果A到B的双向带宽均未指定，则采用默认值 10gb/s（万兆） l 带宽图系统启动时加载内存，基于CREATE/ALTER DATASOURCE命令实时更新 l 静态带宽配置，对CBO起指导性作用（scale but not exact value matters），暂未考虑动态带宽监控

40. 跨DC CBO：计算引擎到数据源的网络带宽 supersql.spark.thriftserver.url.dc1=jdbc:hive2://host1:port1/supersql supersql.hive.thriftserver.url.dc2.cluster1=jdbc:hive2://host2:port2/supersql l多个计算引擎（Spark/Hive ThriftServer2），通过特定的带DataSource前缀名字符串的参数来指定和区分，参数名语义上表示计算引擎所部署的DC/集群 l计算引擎A到数据源B的带宽（写入），定义为带宽图中A的DS前缀名对应节点到数据源名对应节点间的带宽，B到A的带宽同理定义（读取） supersql.webserver.loc=dc1.cluster3 l除外接计算引擎执行外，应用提交的SQL，也可以在SuperSQL Server所在的JVM进程使用 Calcite原生引擎执行，上面的参数定义了SuperSQL Server对应的部署位置 l SuperSQL Server也可以看做一个计算引擎，它与数据源间的带宽同上获得

41. 跨DC CBO：最优引擎挑选最优（单）计算引擎挑选子查询切分，多引擎协同 DC4 Client SuperSQL SuperSQL DC3 DC3 计算引擎 ③ DC2 ② 计算引擎 DC1 Join/Union ① DC1 thive thive hive DC2 ① 对每个数据源下推SQL，基于统计信息估算其输出行数与字节数 ② 对每个计算引擎，计算并行接收所有下推SQL 结果的网络代价 ③ 所有计算引擎CPU代价认为等同，基于网络代价挑选最优引擎执行该查询，这里还需考虑 SuperSQL Server读/写最终查询结果的网络代价计算引擎计算引擎 hive DC1 thive DC2 hive DC2 l 数据源先与本地计算引擎匹配（基于DS前缀名），将计算分到两级的多个计算引擎 l 二级计算引擎看做一级引擎的特殊数据源 l 根据若干因子（如数据量、SQL 模板、用户响应时间或稳定性要求等）挑选合适类型的计算引擎

42. 08 性能评测

43. 测试环境 l l l l l 集群规模：6台服务器硬件：CPU 48核 M10，128GB 内存，3TB 硬盘 OS：TLinux 2.2 64bit Version 2.2 20190320 JDK：RedHat OpenJDK "1.8.0_131” 大数据套件：TBDS 5.0.0.0，基本参数调优 n n n n l l l Spark 2.3.2 – 计算引擎 Hive 2.2.0 – 数据源（TEZ） PostgreSQL 9.2.14 – 数据源 Hadoop 2.7.2 数据集：TPC-DS 1G/100G/1T 查询负载：TPC-DS 99条SQL 对比基线：社区SparkSQL JDBC

44. TPC-DS性能测试结果在相同数据源情况下，数据量越大，SuperSQL的性能优势越明显 TPC-DS 查询数量百分比 TPC-DS 数据集规模 100GB 数据源 SuperSql / Spark Jdbc 时间比值 <= 0.2 SuperSql / Spark Jdbc 时间比值 0.2~0.5 SuperSql / Spark Jdbc 时间比值 0.5~1.0 Hive 98.06% 0.97% 0.97% Hive + PG 混合 85.44% 7.77% 6.80% 所有99条SQL的执行时间（量级几十秒到几十分钟）均值，单源Hive 26倍提升，跨源Hive+PG 5倍提升 TPC-DS 数据集规模 1GB TPC-DS 查询数量百分比数据源 SuperSql / Spark Jdbc 时间比值 <= 0.5 SuperSql / Spark Jdbc 时间比值 0.5~1.0 Hive 43.69% 56.31% PostgreSQL 81.55% 18.45% Hive + PG 混合 38.83% 61.17% 所有99条SQL的执行时间（量级几秒到十几秒）均值，单源Hive 86%提升，单源PG 60%提升，跨源Hive + PG 83%提升

45. 并发子查询性能测试结果并发度 =8~16 时性能最佳 1.8~2.4X 0.3~1.0x 分区越多性能提升越明显 ~ 100X 1~3X hive.fetch.task.conversion.threshold = 1G hive.fetch.task.conversion.threshold = -1

46. 09 系统架构

47. SuperSQL分布式计算引擎需求 l 快速构建：不重复“造轮子”，尽量复用通用开源软件 l 轻量级解耦：不强依赖特定框架，不做或少做侵入性修改 l 场景适配：根据SQL特征，智能挑选主流执行引擎

48. 计算引擎处理流程：数据源下推子计划 ->临时表映射 SuperSQL CBO-> 通知计算引擎 (二次优化) -> 数据源执行 -> 计算引擎汇总 SuperSqlImplementor 计算引擎 select name, cnt EnumerableCalc from view1 join view2 EnumerableJoin on view1.a=view2.a 执行计划 SuperSqlImplementor PostgreSQL Dialect View1 View2 JdbcToEnumerableConverter JdbcToEnumerableConverter JdbcAggregate JdbcJoin JdbcTableScan CREATE TEMPORARY VIEW view1 USING org.apache.spark.sql.jdbc OPTIONS (url ‘jdbc:postgresql://host1:port1/db1’, dbtable ‘(select a, count(*) as cnt from t1 group by a) t’, user “******", password “******"); JdbcTableSca n 数据源 SuperSQL 外接执行引擎要求： 1. 提供 JDBC Client/Server 2. 能够通过 JDBC/HDFS/ 定制 Connector 连接到各类 SuperSQL 数据源 3. 支持动态增加 / 删除数据源或数据源表 SuperSqlImplementor JdbcTableSca n MySQL Dialect CREATE TEMPORARY VIEW view2 USING org.apache.spark.sql.jdbc OPTIONS (url "jdbc:mysql://host2:port2/db2", dbtable ‘(select a, name from t2 join t3 on t2.id = t3.id) t’, user “******", password “******");

49. SuperSQL未来规划 • HBO • 元数据存储重构迁移 • 基于多代价因子的优化 • 对接MPP计算引擎 • 最优执行引擎的智能选择框架

50. Thank you 50

51. 非常感谢您的观看