Spark SQL 原理

Sparksql底层执行原理

Sparksql对sql语句的处理与关系型数据库类似,会有语法或词法的解析、绑定、优化、执行等过程。 Sparksql会先将sql语句解析成抽象语法树AST,然后使用规则Rule对Tree进行绑定、优化等处理。其中Sparksql由Core、Catalyst、Hive、Hive-ThriftServer四部分组成。

• Core:负责数据的输入和输出,如获取数据、查询结果输出成DataFrame等。
• Catalyst:是Sparksql的核心,负责处理整个查询过程,包括解析、绑定、优化等。主要有连个模块,Parse模块和解析模块。
• Hive:负责对Hive数据进行处理。
• Hive-ThriftServer:主要用于对hive的访问。

说明:

1. TreeNode:逻辑计划、表达式等都可以通过Tree来表示,只在内存中维护,并不会进行磁盘持久化,分析器和优化器对树的修改只是替换已有节点。 TreeNode有两个直接子类,QueryPlan和Expression。QueryPlan下又有LogicalPlan和SparkPlan。Expression是表达式体系,不需要执行引擎计算而是可以直接处理或者计算的节点。
2. Rule & RuleExecutor:Rule就是指对逻辑计划要应用的规则,以到达绑定和优化。而RuleExecutor是Rule的实现类,优化器和分析器都需要继承RuleExecutor。每一个子类中都会定义Batch、Once、FixPoint。其中每一个Batch代表着一套规则,Once表示对树进行一次操作,FixPoint表示对树进行多次的迭代操作。RuleExecutor内部提供一个Seq[Batch]属性,里面定义的是RuleExecutor的处理逻辑,具体的处理逻辑由具体的Rule子类实现。

运行原理

1. 使用SessionCatalog保存元数据

   在解析sql语句之前会创建SparkSession,在2.0 版本之前会初始化sqlContext,而SparkSession只是封装了SparkContext和sqlContext的创建,之后会把元数据保存在SessionCatalog中,涉及到表名、字段名称和字段类型。创建临时表或者视图其实就会往SessionCatalog注册。

   说明：SparkSession:Spark新的SQL查询起点,是sqlContext和HiveContext,内部封装了SparkContext,计算实际上是由SparkContext完成的。使用SparkShell时,框架会自动给创建一个名为Spark的SparkSession。

2. 解析sql,使用ANTLR4生成未绑定的逻辑计划

   当调用SparkSession的sql或者sqlContext的sql方法,就会使用Spark的sqlParser进行解析sql。Spark2.0 之前Spark sql使用Scala Parser功能去解析sql表达式,使用类似正则表达式的符号解析,Spark 2.0 之后使用的ANTLR4进行词法解析和语法解析,将sql解析成未绑定的逻辑计划UnresolvedLogicalPlan。

   说明:Antlr4使用了Visitor模式和Listener模式来对解析好的表达式进行处理。Visitor模式:即Antlr4已经将表达式按照你给的规则解析成为一个语法树。然后把语法树的根节点交给用户,并由用户决定该往哪走。Listener模式:即Antlr4已经将表达式按照你给的规则解析成为一个语法树。然后自己执行深度优先遍历,然后在遍历到每个节点的时候给用户发送事件。'注':Sparksql用的就是Visitor方式来遍历节点的!

3. 使用分析器Analyzer绑定逻辑计划

   在该阶段,Analyzer会使用Analyzer Rules并结合SessionCatalog对未绑定的逻辑计划进行解析,生成已绑定的逻辑计划 ResolvedLogicalPlan。

4. 使用优化器Optimizer优化逻辑计划

   优化器也是会定义一套Rules,利用这些Rule对逻辑计划和Exepression进行迭代处理,从而使得树的节点进行合并和优化,得到OptimizedLogicalPlan。

5. 使用SparkPlan生成物理计划

   优化后的逻辑执行计划依然是逻辑的,是不能够被spark系统理解,此时SparkPlan使用Planning Strategies,将优化后的逻辑计划转换成PhysiclPlan。之后使用prepareForException将PhysicalPlan转换成可执行物理计划。

6. 使用execute执行物理计划

   最后使用execute()方法执行可执行物理计划,生成DataFrame。