Spark SQL 查询引擎Catalyst分析

川长思鸟来 2022-02-17 07:55 386阅读 0赞

> ### Spark SQL 查询引擎Catalyst分析 ###

## Catalyst整体架构图 ##

![spark sql æ¥è¯¢å¼æ catalystçæ¶æ][spark sql _ catalyst]

## Catalyst执行流程 ##

catalyst是spark sql的调度核心，遵循传统数据库查询解析步骤，对sql进行解析，转换为逻辑查询计划，物理查询计划，最终转化为Spark的DAG后在执行，下图为Catalyst的执行流程。

*  SqlParser将SQL语句被解析为语法树(AST)，也就是未解析的逻辑查询计划。Parser简单来说是将SQL字符串切分成一个一个Token，再根据一定语义规则解析为一棵语法树。(ANTLR实现)
 *  Analyzer对逻辑查询计划进行属性和关系关联检验，也就是通过定义的一系列规则将未解析的逻辑查询计划借助catalog去解析，如将之前提到的未解析的逻辑查询计划转换成逻辑查询计划。（再次遍历整个语法树，对树上的每个节点进行数据类型绑定以及函数绑定）
 *  Optimizer通过逻辑查询优化将逻辑查询计划转化为优化的逻辑查询计划,优化器是整个Catalyst的核心。下面一篇文章会详细介绍。**包括谓词下推（Predicate Pushdown）、常量累加（Constant Folding）和列值裁剪（Column Pruning）。**
 *  QueryPlanner将逻辑查询计划转换为物理查询计划
 *  prepareForExecution调整数据分布，最后将物理查询计划转换为执行计划进入Spark执行任务

![å¨è¿éæå¥å¾çæè¿°][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTIxMzc0NzM_size_16_color_FFFFFF_t_70]

> ### Spark SQL 优化策略 ###

Spark SQL除了在查询上做了优化同时也在存储上做了优化，下面是sarpk sql的一些优化策略。

### 内存列式存储与内存缓存表 ###

**Spark SQL通过cacheTable将数据存储转换为列式存储，同时将数据加载到内存进行缓存**。**cacheTable相当于在分布式集群的内存物化试图，将数据进行缓存**，这样迭代的或者交互的查询不用在从HDFS读数据，直接从内存读取数据大大减少了I/O开销。列式存储的优势在于Spark SQL只要读取用户需要的列，而不需要想行存储那样需要每次把所有的列读取出来，从而减少了内存缓存的数据量，更高效地利用内存缓存数据，同时减少网络传输和I/O开销。**数据按照列式存储，由于是数据类型相同的数据连续存储，能够利用序列化和压缩减少内存的空间占用。**

### 列式存储压缩 ###

**为了减少内存和磁盘空间占用Spark SQL采用了一些压缩策略对内存列存储数据进行压缩**。Spark SQL支持PassThrough,RunLengthEncoding,IntDelta等多种压缩方式，这样能大幅度减少内存占用，网络开销和I/O开销

### 逻辑查询优化 ###

**Spark SQL在逻辑查询优化上支持列剪枝，谓词下压，属性合并等逻辑查询优化方法**。

列剪枝为了减少不必要的列属性，减少数据传输和计算开销，在查询优化器进行转换的时候会进行列剪枝优化，如下图

![å¨è¿éæå¥å¾çæè¿°][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTIxMzc0NzM_size_16_color_FFFFFF_t_70 1]

例: SELECT Class FROM (SELECT ID, NAME, Class FROM STUDENT) S WHERE S.ID=1  
Catalyst将原来的查询通过谓词下压，选择ID=1先执行，这样过滤掉了大部分数据，通过属性合并将最后的投影只做一次最终保留Class属性

### Join优化 ###

Spark SQL借鉴了传统数据库查询优化的精髓，并在分布式环境下进行特定的优化策略调整和创新。Spark SQL对join进行了优化并支持多种连接算法，例如: BroadcastHashJoin,BroadcastNestedLoopJoin,HashJoin,LeftSemiJoin等

BroadcastHashJoin将小表转换为广播变量进行广播，避免shuffle开销，最后在分区内进行Hash连接。这里用的就是hive中的Map Side Join 思想。同时用了DBMS中的Hash连接算法做连接。

原文：https://blog.csdn.net/u012137473/article/details/84995896   
原文：https://blog.csdn.net/u012137473/article/details/84980543

[spark sql _ catalyst]: /images/20220217/3f6b2e783bdf416f85d66eb14d4fab5a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTIxMzc0NzM_size_16_color_FFFFFF_t_70]: /images/20220217/7d7c1013608e4b929b343fb9c93836fb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTIxMzc0NzM_size_16_color_FFFFFF_t_70 1]: /images/20220217/af16380a9bb74c1f8507741bcd446ef3.png