spark SQL优化器catalyst学习

一、Catalyst 概述

1. Catalyst 是 Spark SQL 的优化器，它负责将 SQL 查询转换为物理执行计划。Catalyst 优化器的目标是生成高效的执行计划，以最小化查询的执行时间。它使用了多种优化技术，包括基于规则的优化、基于代价的优化和动态规划等。
2. 我们写的SQL语句，会经过一个优化器(catalyst)，转化为RDD，交给集群执行。
   select * from table_a
3. 语法。select * table_a
4. 词法。selectS * from table_a
5. AST。abstract syntax tree / 抽象语法树 / 语法树 / syntax tree
   暂时无法在飞书文档外展示此内容
6. 解析引擎：负责将SQL解析成task。Catalyst的地位类似于Calcite(负责Hive SQL解析优化)，Spark SQL独有的Catalyst，解析优化。
7. MySQL
8. Hive SQL
9. Spark SQL
10. Flink SQL
11. Doris
12. Presto
13. 计算引擎：task是谁来执行。
14. MySQL
15. MapReduce
16. Spark
17. Tez
18. Flink

• Spark on Hive和 Hive on spark的区别？
  • Hive on Spark：Hive是Hive SQL，解析引擎；Spark是计算引擎。
  • Spark on Hive：
    • Hive
      1. 代表是数仓(常见的分层)；
      2. 理解：Hive SQL，解析引擎。
    • Spark：代表解析引擎、计算引擎。
    • 含义：通过spark SQL做数仓，层与层的转换。
• SQL到RDD中间经过了一个Catalyst，它就是SparkSQL的核心,是计对 Spark SQL语句执行过程中的查询优化框架，基于Scala函数式编程结构。
• RDD的运行流程:RDD->DAGScheduler ->TaskScheduler->worrker，任务会按照代码所示运行，依赖开发者的优化，开发者的会在很大程度上影响运行效率。而SparkSQL的Dataset和SQL并不是直接生成计划交给集群执行，而是经过Catalyst的优化器，这个优化器能够自动帮助开发者优化代码
• 我们要了解SparkSQL的执行流程,那么理解Catalyst的工作流程是非常有必要的。
  二、 Catalyst 的优化过程
  暂时无法在飞书文档外展示此内容
  Catalyst 的优化过程大致可以分为以下几个阶段：

1. 解析 ( Parsing )：将 SQL 查询解析为抽象语法树 ( AST )。parser模块目前都是使用第三方类库ANTLR进行实现的。在这个过程汇总，会判断SQL语句是否符合规范，比如select from where等这些关键字是否写对。
   暂时无法在飞书文档外展示此内容
2. 分析 ( Analysis )：对 AST 进行语义分析，检查查询的合法性和完整性。该模块会遍历整个AST，并对AST上的每个节点进行数据类型绑定以及函数绑定，然后根据源数据信息系catelog对数据表中的字段进行解析，此过程会判断SQL语句的表名，字段名是否真的在元数据库里存在。元数据信息主要包括两部分：表的scheme和基本函数信息。
3. 表的scheme：
   1. 基本定义。列名，数据类型。
   2. 表的数据格式。json、text
   3. 表的物理位置。
4. 基本函数
   暂时无法在飞书文档外展示此内容
5. 优化 ( Optimization )：应用各种优化规则和策略，生成不同的执行计划。主要分为RBO和CBO两种优化策略，其中RBO(Rule-Based Optimizer)是基于规则优化，CBO(Cost-Based Optimizer)是基于代价优化。常见的规则有：

• 谓词下推predicate Pushdown：将过滤操作下推到join之前进行，之后在进行join的时候，数据量将会显著的减少，join耗时必然降低。
  暂时无法在飞书文档外展示此内容

select
    *
from 
    table1
inner join
    table2
on table1.id = table2.id
where 
    table1.age > 20
    and table2.cid = 1
1
2
3
4
5
6
7
8
9
10

上面的语句会自动优化为如下所示：

select
    *
from 
(
    select 
        *
    from
        table1
    where 
        table1.age > 20
)    a
inner join
(
    select 
        *
    from
        table2
    where 
        table2.cid = 1
)    b
on a.id = b.id
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

即在资产许那阶段就提前将数据进行过滤，后续的join和shuffle数据量会大大减少。

• 列值裁剪column pruning：在谓词下推后，可以把表中没有用到的列裁剪掉，这一优化一方面大幅度减少了网络，内存的数据量消耗，另一方面对于劣势存储数据库来说大大提高了扫描效率。

select
    a.name
    ,a.age
    ,b.cid
from 
(
    select 
        *
    from
        table1
    where 
        table1.age > 20
)    a
inner join
(
    select 
        *
    from
        table2
    where 
        table2.cid = 1
)    b
on a.id = b.id
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

上面的语句会自动优化如下图所示：

select
    a.name
    ,a.age
    ,b.cid
from 
(
    select 
        name
        ,age
    from
        table1
    where 
        table1.age > 20
)    a
inner join
(
    select 
        cid
    from
        table2
    where 
        table2.cid = 1
)    b
on a.id = b.id
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

就是提前将需要的列查询出来，其他不需要的列裁剪掉。

• 常量累加 constant folding：比如计算 x + (100 + 80) -> x + 180，虽然是一个很小的改动，但是意义巨大。如果没有进行优化，每一条结果都需要执行一次100 + 80的操作，然后再与结果相加，优化后就不需要再次执行100 + 80的操作。

select 
    1 + 1 id
from
    table1
1
2
3
4

上面的语句会自动优化如下图所示：

select 
    2 id
from
    table1
1
2
3
4

就是会提前将1 + 1计算成2，再赋给id列的每行，不用每次都计算一次1+1
4. SparkPlanner模块：

1. 将优化后的逻辑执行计划(OptimizedLogicalPlan)转换成Physical Plan(物理计划)，也就是Spark可以真正的执行的计划。比如join算子，Spark根据不同常见为该算子制定了不同的算法策略，有BroadCastHashJoin,ShuffleHashJoin以及SortMergeJoin等，物理执行假话实际上就是在这些具体实现中挑选一个耗时最小的算法实现。
2. 具体的实现手段：
   1. SparkPlanner对优化后的逻辑计划进行换算，是生成了多个可以执行的物理计划Physical Plan；接着CBO(基于代价优化)优化策略或根据Cost Model算出每个Physical Plan的代价，并选取最小代价的Physical Plan作最终的Physical Plan。
   2. CostmModel模块：主要根据过去的性能统计数据，选择最佳的物理执行计划，这个过程的优化就是CBO(基于代价优化)
   备注：以上2、3、4步骤结合起来，就是catalyst优化器。
3. 执行物理计划：最后一句最优的物理执行计划，生成Java字节码，将SQL转化为DAG，以RDD的形式进行操作
4. 选择 ( Selection )：根据代价模型选择最优的执行计划。
5. 代码生成 ( Code Generation )：将优化后的执行计划转换为 Spark 代码。
   三、 Catalyst 的优化规则
   Catalyst 提供了许多优化规则，用于改进查询的执行计划。以下是一些常见的优化规则：
6. 列剪枝 ( Column Pruning )：删除不必要的列，减少数据传输。
7. 分区剪枝 ( Partition Pruning )：根据分区过滤条件，只读取必要的分区。
8. 谓词下推 ( Predicate Pushdown )：将过滤条件尽可能地向下推送到数据源，减少数据的读取量。
9. 聚合优化 ( Aggregation Optimization )：合并相同的聚合操作，避免重复计算。
10. 连接优化 ( Join Optimization )：选择合适的连接算法，优化连接操作。
    四、 Catalyst 的代价模型
    Catalyst 采用了基于规则的代价模型来评估执行计划的代价。代价模型考虑了以下因素：
11. 数据量 ( Data Volume )：表的数据大小和分区数。
12. 计算资源 ( Compute Resources )：CPU、内存和网络带宽等。
13. I/O 开销 ( I/O Overhead )：数据读取和写入的开销。
14. 数据倾斜 ( Data Skew )：数据分布不均衡导致的性能问题。
    五、 Catalyst 的代码生成
    Catalyst 将优化后的执行计划转换为 Spark 代码，包括RDD操作和SQL表达式。代码生成过程使用了模板和宏来实现代码的重用和简洁性。
    六、总结
    Spark SQL 的 Catalyst 优化器是一个强大而灵活的优化框架，它采用了多种优化技术和策略，以生成高效的执行计划。了解 Catalyst 的优化过程和规则，可以帮助我们更好地编写高效的 Spark SQL 查询。

---

以上是一个关于 Spark SQL 优化器 Catalyst 的学习文档，希望对你有所帮助。如果你有任何问题或建议，请随时与我交流。