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 技术交流社区」，获取更多技术干货 ▼
版权归原作者 Apache Spark中国社区 所有， 如有侵权，请联系我们删除。