Paimon 与 Spark 的集成(二):查询优化

Paimon

Apache Paimon (incubating) 是一项流式数据湖存储技术，可以为用户提供高吞吐、低延迟的数据摄入、流式订阅以及实时查询能力。Paimon 采用开放的数据格式和技术理念，可以与 Flink / Spark / Trino 等诸多业界主流计算引擎进行对接，共同推进 Streaming Lakehouse 架构的普及和发展。

Paimon x Spark

‍‍Apache Spark，作为大数据处理的统一计算分析引擎，不仅支持多种语言的高级API使用，也支持了丰富的大数据场景应用，包括结构化数据处理的Spark SQL、用于机器学习的MLlib，用于图形处理的GraphX，以及用于增量计算和流处理的Structured Streaming。Spark已经成为了大数据领域软件栈中必不可少的组成部分。对 Paimon 来说，为了在准实时和离线湖仓场景更加便利的落地，与 Spark 深度、全面的集成势在必行。

在之前的Paimon Release版本，我们着重丰富Paimon在功能上和Spark SQL生态的集成，包括Schema Evolution，Structured Streaming Read/Write，Dynamic Insert Overwrite Partition，Update/Merge Into等等。在最近发布的0.6和0.7版本，我们开始在Paimon基于Spark SQL查询性能上做一些工作。在初期我们会结合Spark SQL已有的优化规则和框架，让Paimon充分利用到这些。通过一系列优化，我们将 Paimon x Spark 在 TpcDS 上的性能提高了37+%，已基本和 Parquet x Spark 持平。下文将对其中的关键优化点进行详细介绍。‍‍

动态分区裁剪

动态分区裁剪（Dynamic Partition Prunning，DPP）在SQL优化中是常见的优化点，本质上是谓词下推（Predicate PushDown）的一种拓展，其目的是最小化从数据源中读取数据的IO成本，也进而减少了计算成本。

在数仓中，常常将较大的事实表和很小的维度表关联查询，且事实表需要根据维表中的字段信息来进行过滤，如下面TpcDS Q14中的SQL片段：

        
          select ss_quantity quantity ,ss_list_price list_price
        
        
          from store_sales, date_dim
        
        
          where ss_sold_date_sk = d_date_sk and d_year between 1999 and 1999 + 2
        
        
          order by quantity limit 10;
        
      

在不支持DPP的情况下的执行计划简化如下：

Paimon应用的是Spark DataSource V2的查询框架，该框架在Spark3.2后提供了 SupportsRuntimeFiltering 接口用于V2表实现运行时的动态过滤。理论上，任何字段（包括普通数据字段和分区字段）的过滤条件都能被应用，但一般而言仅分区字段的过滤条件能够被完全应用，即无需上层的Filter的节点再使用该过滤条件去选择数据。Paimon表通过该接口实现了动态分区裁剪的能力。在支持DPP后执行计划如下所示：

在1T的TpcDS数据集下，应用DPP后  store_sales  表参与join的数据量从27亿 减少到16亿。仅应用到该优化后，Q14运行时间减少到原来的～55%，1T TpcDS数据集的查询性能整体提升20+%；
相关代码： https://github.com/apache/incubator-paimon/pull/2411 https://github.com/apache/incubator-paimon/pull/2421

Exchange复用

Exchange是Spark中物理计划中一个关键的操作，对应逻辑计划中的Shuffle。在执行阶段，Exchange可以代表某个SQL中部分Plan输出的数据。在复杂的SQL中，我们可以通过公共表表达式（Common Table Expression，CTE）语法定义一个SQL片段，用于简化整个SQL或者被多次使用。以下面简化的TpcDS Q23为例，定义的其中一个CTE frequent_ss_items 在整个SQL中被两次使用。

        
          with frequent_ss_items as (
        
        
            select substr(i_item_desc,1,30) itemdesc, i_item_sk item_sk, d_date solddate, count(*) cnt
        
        
            from store_sales, date_dim, item
        
        
            where ss_sold_date_sk = d_date_sk and ss_item_sk = i_item_sk and d_year in (2000,2000+1,2000+2,2000+3)
        
        
            group by substr(i_item_desc,1,30),i_item_sk,d_date
        
        
            having count(*) >4
        
        
          ),
        
        
          max_store_sales as (...),
        
        
          best_ss_customer as (...)
        
        
          select sum(sales)
        
        
          from (
        
        
            select cs_quantity*cs_list_price sales
        
        
                 from catalog_sales
        
        
                     ,date_dim
        
        
                 where d_year = 2000
        
        
                   and d_moy = 2
        
        
                   and cs_sold_date_sk = d_date_sk
        
        
                   and cs_item_sk in (select item_sk from frequent_ss_items)
        
        
                   and cs_bill_customer_sk in (select c_customer_sk from best_ss_customer)
        
        
                union all
        
        
                select ws_quantity*ws_list_price sales
        
        
                 from web_sales
        
        
                     ,date_dim
        
        
                 where d_year = 2000
        
        
                   and d_moy = 2
        
        
                   and ws_sold_date_sk = d_date_sk
        
        
                   and ws_item_sk in (select item_sk from frequent_ss_items)
        
        
                   and ws_bill_customer_sk in (select c_customer_sk from best_ss_customer)
        
        
          ) y
        
        
          limit 100;
        
      

显然在执行阶段，我们希望 frequent_ss_items 仅被执行一次，执行后的数据可以缓存，然后分别执行后续和 catalog_sales 以及 web_sales 表的Join操作。针对这个场景，Spark提供了Exchange复用的优化，期待的执行计划简化如下所示：

但该优化依赖算子Plan中各个物理操作的 hashCode 来确定实际运行时是否可以复用。我们定位并解决了Paimon中存在的实现问题，使得Paimon可以使用到Spark提供的Exchange复用的优化，从而减少不必要的冗余计算，也降低了IO和网络的开销。仅应用到该优化后，Q23运行时间减少到原来的～50%，1T TpcDS数据集的查询性能整体提升13+%；

相关代码：

https://github.com/apache/incubator-paimon/pull/2488

动态调整Scan并发

任务实际执行时的并发度是影响作业运行性能的关键之一。Spark提供了 spark.sql.shuffle.partitions 参数来调整Join或者Agg等算子的并发，也提供了自适应查询执行（Adative Query Execution，AQE）框架动态调整并发，但这些都无法影响到读取数据源Scan阶段的并发。

在DataSource V2的框架下，数据源的Scan方式包括并发完全由DataSource自己决定。我们以TpcDS Q19为例：

        
          select  i_brand_id brand_id, i_brand brand, i_manufact_id, i_manufact,
        
        
             sum(ss_ext_sales_price) ext_price
        
        
           from date_dim, store_sales, item,customer,customer_address,store
        
        
           where d_date_sk = ss_sold_date_sk
        
        
             and ss_item_sk = i_item_sk
        
        
             and i_manager_id=8
        
        
             and d_moy=11
        
        
             and d_year=1998
        
        
             and ss_customer_sk = c_customer_sk 
        
        
             and c_current_addr_sk = ca_address_sk
        
        
             and substr(ca_zip,1,5) <> substr(s_zip,1,5) 
        
        
             and ss_store_sk = s_store_sk 
        
        
           group by i_brand
        
        
                ,i_brand_id
        
        
                ,i_manufact_id
        
        
                ,i_manufact
        
        
           order by ext_price desc
        
        
                   ,i_brand
        
        
                   ,i_brand_id
        
        
                   ,i_manufact_id
        
        
                   ,i_manufact
        
        
          limit 100 ;
        
      

其中 customer_address 和 store 基于 substr(ca_zip,1,5) <> substr(s_zip,1,5) 条件Join。

在未引入CBO对join重排序的情况下，这两张表通过BroadcastNestedLoopJoin来实现，没有引入Exchange调整Join的并发。执行计划如下图所示：

在未引入优化之前，由于 customer_address 表的数据分片较小，但任务计算负载较高（数据Join后严重膨胀），整体执行性能很差。

Paimon根据这种问题提供了基于当前作业的可用core数来动态调整数据源的数据分片的能力，也进而调整并发，从而提升查询效率。 仅应用该优化后，Q19运行时间减少到原来的～25%，1T TpcDS数据集的查询性能整体提升14+%；

相关代码： https://github.com/apache/incubator-paimon/pull/2482

合并标量子查询

类似于Exchange复用，合并标量子查询优化会遍历整个SQL逻辑执行计划，提取出标量子查询（ScalarSubQuery），尝试将多个标量子查询合并起来，使得仅执行一次子查询得到多个标量值。

我们以TpcDS Q9的片段为例，整个Q9由5个case-when语句构成。

        
          select case when (select count(*) 
        
        
                            from store_sales 
        
        
                            where ss_quantity between 1 and 20) > 74129
        
        
                      then (select avg(ss_ext_discount_amt) 
        
        
                            from store_sales 
        
        
                            where ss_quantity between 1 and 20) 
        
        
                      else (select avg(ss_net_paid)
        
        
                            from store_sales
        
        
                            where ss_quantity between 1 and 20) end bucket1
        
        
          from reason
        
        
          where r_reason_sk = 1;
        
      

在该SQL中， case when 的条件， then 和 else 语句三个部分使用同样的过滤条件读取同一张表，仅聚合表达式不同。在没有应用到这个优化的情况下，执行计划如下所示：

Spark本身提供了 MergeScalarSubQueries 的优化规则，但从实现上没法更好的对接到Paimon这样的DataSource V2表，因此我们在Paimon侧单独实现，并通过Spark提供的Extensions的接口将Paimon自实现的优化注入到了Spark优化器中。在应用该优化后，执行计划如下所示：

由此可见，合并标量子查询优化有效的减少了冗余的计算，提升了Paimon在该场景下的查询性能。仅应用该优化后，Q9运行时间减少到原来的～57%。
相关代码： https://github.com/apache/incubator-paimon/pull/2657

Cost-Based优化

Spark SQL允许使用基于成本的优化（Cost-Based Optimizer，CBO）来提升查询性能，主要用于多路Join的场景，使用动态规划算法来选择Cost最低的Join顺序。要想使得这个优化能更有效，依赖于计算Cost的模型，以及表的表级和列级统计信息的收集，而其中列级统计信息在评估Plan算子节点的运行时统计信息中尤为重要。

新版本的Paimon在元数据中增加了statistics的信息，可以通过原生的Spark Analyze命令完成收集，并对接到了Spark SQL，使得Spark SQL可以利用Paimon的表级/列级信息进行查询优化。我们以TpcDS Q24a为例：

        
          with ssales as
        
        
          (select c_last_name, c_first_name, s_store_name, ca_state, s_state, i_color, i_current_price, i_manager_id, i_units, i_size, sum(ss_net_paid) netpaid
        
        
          from store_sales, store_returns, store, item, customer, customer_address
        
        
          where ss_ticket_number = sr_ticket_number
        
        
            and ss_item_sk = sr_item_sk
        
        
            and ss_customer_sk = c_customer_sk
        
        
            and ss_item_sk = i_item_sk
        
        
            and ss_store_sk = s_store_sk
        
        
            and c_current_addr_sk = ca_address_sk
        
        
            and c_birth_country <> upper(ca_country)
        
        
            and s_zip = ca_zip
        
        
          and s_market_id=8
        
        
          group by c_last_name, c_first_name, s_store_name, ca_state, s_state, i_color, i_current_price, i_manager_id, i_units, i_size)
        
        
          select c_last_name, c_first_name, s_store_name, sum(netpaid) paid
        
        
          from ssales
        
        
          where i_color = 'peach'
        
        
          group by c_last_name, c_first_name, s_store_name
        
        
          having sum(netpaid) > (select 0.05 * avg(netpaid) from ssales)
        
        
          order by c_last_name, c_first_name, s_store_name;
        
      

其中CTE ssales 的部分，仅提供表级统计信息的情况下的执行计划大致如下所示，包括两个SortMergeJoin，其中左侧虚线框更是大数据量间的Join操作，严重影响性能。

而执行Analyze提供了列级统计信息后执行计划大致如下所示，对参与Join的表进行了重排序，且所有Join都是BroadcastHashJoin的方式执行。

Paimon提供了完整的statistics，借助于CBO框架，不仅可以提升相应的查询性能，也可以使得在正常资源配置下无法跑通的SQL能够正常运行，比如TpcDS Q72。在之前优化项基础上叠加应用该优化后，Q24运行时间减少到原来的～23%，1T TpcDS数据集的查询性能整体提升~30%；
相关代码： https://github.com/apache/incubator-paimon/pull/2677 https://github.com/apache/incubator-paimon/pull/2752

https://github.com/apache/incubator-paimon/pull/2798

优 化 效 果

本文使用阿里云 EMR 5.16.0版本，集群节点的属性如下：

• master: 1 * ecs.g7.8xlarge 32 vCPU 128 GiB

• core: 6 * ecs.g7.8xlarge 32 vCPU 128 GiB

使用的组件及版本如下：

• Paimon: 0.8-SNAPSHOT （对应到commit：193df7345aa520f8b45125cdd85588a91a3fc3a9）

• Spark: 3.3.1 (额外cherry-pick SPARK-41378 ，以支持DataSource V2下的stats相关功能）

启用的Spark相关配置：

spark.executor.cores	4
spark.executor.memory	14g
spark.executor.memoryOverhead	2g
spark.dynamicAllocation.enabled	true
spark.sql.cbo.enabled	true
spark.sql.cbo.joinReorder.enabled	true
spark.sql.autoBroadcastJoinThreshold	128m

Paimon表选用append表（无主键表），使用parquet作为文件格式，设置bucket=-1（最新代码已经默认设置： PAIMON-2829 ），这样便于和Spark parquet表进行对比。

上图为我们使用parquet表（带有表级统计信息，即rowCount和sizeInByte两个指标）作为基准，以此向右分别为优化前和应用这些优化后的Paimon表（仅带表统计信息），以及Parquet表和Paimon表在收集到Column级别统计信息时的查询较基准的性能对比。 对比可见，在一般情况下（无column级统计信息）优化后的Paimon和Parquet已经基本持平。开启column级统计信息后，Paimon较Parquet慢~8%，这中间的差距也将是性能优化继续跟进的方向之一。

后 续 规 划 

在湖仓体系下，我们认为读写查询优化一直是一项任重而道远的事情。当前的优化主要集中在让Paimon充分利用到Spark SQL现有的优化规则或者优化框架。在继续推进的同时，我们也会利用Paimon自身的特性，比如Index或者Clustering等，以及优化Scan等进一步提升Paimon性能。

另外，在当前湖仓场景下，依然有很多无主键表的使用，后续对append表支持Upsert能力也是重要的规划之一。

< 往 期 精 彩 推 荐 >

---

▼ 关注「 Apache Spark 技术交流社区 」，获取更多技术干货  ▼