如何进行EMR Spark Relational Cache的执行计划重写

这篇文章将为大家详细讲解有关如何进行EMR spark Relational Cache的执行计划重写，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。

背景

EMR Spark提供的Relational Cache功能，可以通过对数据模型进行预计算和高效地存储，加速Spark sql，为客户实现利用Spark SQL对海量数据进行即时查询的目的。Relational Cache的工作原理类似物化视图，在用户提交SQL语句时对语句进行分析，并选出可用的预计算结果来加速查询。为了实现高效地预计算结果复用，我们构建的预计算缓存一般都较为通用，因此对于用户query，还需进行进一步的计算方能获得最终结果。因此，如何快速地找出匹配的缓存，并构建出准确的新执行计划，就显得尤为重要。

在Hive 3.x中支持的Materialized View，利用了Apache Calcite对执行计划进行重写。考虑到Spark SQL使用Catalyst进行执行计划优化，引入Calcite太重，因此EMR Spark中的Relational Cache实现了自己的Catalyst规则，用于重写执行计划。下面将介绍执行计划重写的相关内容。

执行计划重写

准备工作

Spark会把用户查询语句进行解析，依次转化为Unresolved Logical Plan（未绑定的逻辑计划）、Resolved Logical Plan（绑定的逻辑计划）、Optimized Logical Plan（优化的逻辑计划）、Physical Plan（物理计划）。其中，未优化的逻辑计划根据用户查询语句不同，会有较大区别，而Relational Cache作为优化的一部分，放在逻辑计划优化过程中也较为合适，因此我们拿到的用户查询计划会是优化中的逻辑计划。要与优化中的逻辑计划匹配，我们选择把这个重写过程放在Spark优化器比较靠后的步骤中，同时，预先将Relational Cache的逻辑计划进行解析，获得优化后的Cache计划，减小匹配时的复杂程度。这样，我们只需匹配做完了谓词下推、谓词合并等等优化之后的两个逻辑计划。

基本过程

在匹配时，我们希望能尽可能多得匹配计算和IO操作，因此，我们对目标计划进行前序遍历，依次进行匹配，尝试找到最多的匹配节点。而在判断两个节点是否匹配时，我们采用后序遍历的方式，希望尽快发现不匹配的情况，减少计划匹配的执行时间。然后我们会根据匹配结果，对计划进行重写，包括对于Cache数据进行进一步的Filter、Project、Sort甚至Aggregate等操作，使其与匹配节点完全等价，然后更新逻辑计划节点的引用绑定，无缝替换到逻辑计划中，这样就能轻松获得最终的重写后的计划。

Join匹配

Spark中的Join都是二元操作，而实际的Join顺序可能根据一些策略会有很大区别，因此对于Join节点，必须进行特殊处理。我们会首先将逻辑计划进行处理，根据缓存计划的Join顺序进行Join重排。这一步在树状匹配之前就进行了，避免不断重复Join重排带来的时间浪费。重排后的Join可以更大概率地被我们匹配到。

为了实现Cache的通用性，根据星型数据模型的特点，我们引入了Record Preserve的概念。这和传统数据库中的Primary Key/Foreign Key的关系较为类似，当有主键的表与非空外键指向的表在外键上进行Join时，记录的条数不会变化，不会膨胀某条记录，也不会丢失某条记录。PK/FK的语意在大数据处理框架中经常缺失，我们引入了新的DDL让用户自定义Record Preserve Join的关系。当用户定义A Inner Join B是对于A表Record Preserve时，我们也会把A Inner Join B和A的关系匹配起来。有了PK/FK的帮助，我们能匹配上的情况大大增加了，一个Relational Cache可以被更多看似区别巨大的查询共享，这可以很好的为用户节约额外的存储开销和预计算开销。

Aggregate匹配

一般的Aggregate匹配较为简单，而Spark支持的Grouping Set操作，会构建出Expand逻辑计划节点，相当于把一条记录转为多条，使用Grouping ID进行标记。由于Expand的子节点是所有Grouping的情况共用的，这里我们只对子节点进行一次匹配，再分别进行上面的Grouping属性和Aggregate属性的匹配。主要是验证目标聚合所需的属性或者聚合函数都能从某个Grouping ID对应的聚合结果中计算出来，比如粗粒度的Sum可以对细粒度的Sum进行二次Sum求和，而粗粒度的Count对细粒度的Count也应通过二次Sum求和，粗粒度的Average无法仅从细粒度的Average中还原出来等等。

计划重写

找出匹配的逻辑计划之后，就是重写逻辑计划的过程。对于无需二次聚合的逻辑计划，直接根据缓存数据的schema，从缓存数据的Relation中选择所需列，根据条件过滤后，进行后续操作。如果还需二次聚合，选择所需列时需保留外部要用的所有列，以及聚合时需要的列，还有聚合函数需要的数据。二次聚合的聚合函数需要根据实际情况进行重写，确保能使用Relational Cache中已经初步聚合的结果。这里面需要根据聚合的语意判断是否能够二次聚合。如果时Grouping Set的聚合，二次聚合之前还需选择正确的Grouping ID进行过滤。经过二次聚合后，步骤大体和普通的重写一致，只需替换到目标计划中即可。

结果

我们以一个例子来具体说明逻辑计划的重写结果。Star Schema Benchmark（论文链接https://www.cs.umb.edu/~poneil/StarSchemaB.pdf）是星型模型数据分析的一个标准Benchmark，其结构定义如图所示：

我们构建Relational Cache的SQL语句如下：

SELECT GROUPING_ID() AS grouping_id, lo_discount, s_city, c_city, p_cateGory, d_year, lo_quantity, d_weeknuminyear, s_nation, s_region, p_mfgr, c_region, d_yearmonth, p_brand, c_nation, d_yearmonthnum, SUM(lo_revenue) AS lo_revenue_SUM, SUM(lo_supplycost) AS lo_supplycost_SUM, SUM(V_REVENUE) AS V_REVENUE_SUMFROM supplier, p_lineorder, dates, customer, partWHERE lo_orderdate = d_datekey AND lo_cusTKEy = c_custkey AND lo_suppkey = s_suppkey AND lo_partkey = p_partkeyGROUP BY lo_discount, s_city, c_city, p_category, d_year, lo_quantity, d_weeknuminyear, s_nation, s_region, p_mfgr, c_region, d_yearmonth, p_brand, c_nation, d_yearmonthnum GROUPING SETS ((d_year, d_weeknuminyear, lo_discount, lo_quantity), (d_year, lo_discount, lo_quantity), (lo_discount, lo_quantity), (d_yearmonthnum, lo_discount, lo_quantity), (d_year, p_category, p_brand, s_region), (d_year, p_category, s_region), (d_year, s_region), (d_year, s_region, c_region, s_nation, c_nation), (d_year, s_city, c_city, s_nation, c_nation), (d_year, s_city, c_city), (d_year, d_yearmonth, s_city, c_city), (d_year, s_region, c_region, c_nation, p_mfgr), (d_year, s_region, s_nation, c_region, p_mfgr, p_category), (d_year, s_nation, s_city, c_region, p_brand, p_category, p_brand), (d_year, s_nation, s_city, c_region, p_brand, p_category), (d_year, s_nation, s_city, c_region, p_category, p_brand))

我们从中选出一条查询作为示例。具体查询语句：

select c_city, s_city, d_year, sum(lo_revenue) as revenue    from customer, lineorder, supplier, dates    where lo_custkey = c_custkey        and lo_suppkey = s_suppkey        and lo_orderdate = d_datekey        and c_nation = 'UNITED KINGDOM'        and (c_city='UNITED KI1' or c_city='UNITED KI5')        and (s_city='UNITED KI1' or s_city='UNITED KI5')        and s_nation = 'UNITED KINGDOM'        and d_yearmonth = 'Dec1997'    group by c_city, s_city, d_year    order by d_year asc, revenue desc

原始逻辑计划如下所示：

Sort [d_year#47 ASC NULLS FIRST, revenue#558L DESC NULLS LAST], true+- Aggregate [c_city#22, s_city#39, d_year#47], [c_city#22, s_city#39, d_year#47, sum(cast(lo_revenue_SUM#773L as bigint)) AS revenue#558L]   +- Filter ((((((((isnotnull(s_nation#40) && ((s_city#39 = UNITED KI1) || (s_city#39 = UNITED KI5))) && (s_nation#40 = UNITED KINGDOM)) && isnotnull(d_yearmonth#49)) && (d_yearmonth#49 = Dec1997)) && isnotnull(c_nation#23)) && (c_nation#23 = UNITED KINGDOM)) && ((c_city#22 = UNITED KI1) || (c_city#22 = UNITED KI5))) && (grouping_id#662 = 19322))      +- Relation[grouping_id#662,lo_discount#759,s_city#39,c_city#22,p_category#762,lo_quantity#763,d_weeknuminyear#764,s_nation#40,s_region#766,p_mfgr#767,c_region#768,p_brand1#769,c_nation#23,d_yearmonthnum#771,d_yearmonth#49,lo_revenue_SUM#773L,lo_supplycost_SUM#774L,V_REVENUE_SUM#775L,d_year#47] parquet

由此可见，执行计划大大简化，我们可以做到亚秒级响应用户的命中查询。

进一步优化

在实际测试过程中，我们发现当多个Relational Cache存在时，匹配时间线性增长明显。由于我们在metastore中存储的是Cache的SQL语句，取SQL语句和再次解析的时间都不容小觑，这就使得匹配过程明显增长，背离了我们追求亚秒级响应的初衷。因此我们在Spark中构建了逻辑计划缓存，将解析过的Relational Cache的计划缓存在内存中，每个Relational Cache只缓存一份，计划本身占用空间有限，因此我们可以缓存住几乎所有的Relational Cache的优化后的逻辑计划，从而在第一次查询之后，所有查询都不再收到取SQL语句和再次解析的延迟困扰。经过这样的优化，匹配时间大幅减少到100ms的量级。

关于如何进行EMR Spark Relational Cache的执行计划重写就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。