背景

Join是一個非常耗費資源耗費時間的操作，特別是數據量很大的情況下。一般流程上會涉及底層表的掃描/shuffle/Join等過程, 如果我們能夠盡可能的在靠近源頭上減少參與計算的數據，一方面可以提高查詢性能，另一方面也可以減少資源的消耗(網絡/IO/CPU等)，在同樣的資源的情況下可以支撐更多的查詢。

目前在SparkSQL中有Filter下推優化，包括兩個維度:

生成Filter

SparkSQL會從用戶的SQL語句中獲取到Filter

• 直接顯示獲取

  select * from A where a=1

  生成Filter(a=1) on A

• 隱式推斷

  select * from A, B where A.a = B.b and A.a=1

  推斷出Filter(b=1) on B

Filter優化

利用生成的Filter算子可以優化，比如:

• 將Filter盡量下推到靠近DataSource端

• 如果Filter中的列是分區列，可以提前對DataSource進行分區裁剪，只掃描需要的分區數據

Runtime Filter是針對Equi-Join場景提出的一種新的生成Filter的方式，通過動態獲取Filter內容來做相關優化。

Runtime Filter原理

優化對象

Equi Join, 形如

select x,y from A join B on A.a = B.b

其中A是一個小表(如維表)，B是一個大表(如事實表)
備注:?A/B也可以是一個簡單的子查詢

優化思路

如上述小表A和大表B進行Join，Join條件為A.a=B.b，實際Join過程中需要對大表進行全表掃描才能完成Join操作，極端情況下如A.a僅僅只有一條記錄，也需要對B表全表掃描，影響性能。

如果在B表掃描之前，能獲取A表的a的相關信息(如所有的a值，或者a的min/max/Bloomfilter等統計信息)，并在實際執行Join之前將這些信息對B表的數據進行過濾，而不是全表掃描，可以大大提高性能。

兩種場景

根據大表B參與join的key(b)的屬性，可以分別采集小表A參與join的key(a)的信息:

b是分區列

如上b為大表B的一個分區列，則可以提前收集A.a列的所有值，然后利用A.a的值對B表的b列進行分區裁剪

b不是分區列

不能做分區裁剪，只能在實際數據掃描的過程中進行過濾。可以提前收集A.a列的min/max/Bloomfilter的統計信息，然后利用這些統計信息對B表進行數據過濾，這個過濾又可以分成兩種粒度:

• 可下推到存儲層，減少數據掃描
  如底層文件格式是Parquet/ORC, 可以將相關過濾謂詞(min/max等)下推到存儲層面，從而減少實際掃描的數據。
• 掃描后數據過濾
  不能下推到存儲層的，可以在數據被掃描后做條件過濾，減少后續參與計算的數據量(如shuffle/join等)

Runtime Filter實現

Runtime Filter的實現主要在Catalyst中，分為4個步驟:

謂詞合成

在用戶SQL生成的邏輯執行計劃樹(logical plan)中，尋找滿足條件的Equi-Join節點，然后根據上面的思路，在Join的大表B側插入一個新的Filter節點，如Filter(In(b, Seq(DynamicValue(a, A))), B)

謂詞下推

上面生成的新的Filter會經過PushDownPredicate的Rule，盡量下推靠近DataSource附近

物理執行計劃生成

該階段會將上面下推的Filter(In(b, Seq(DynamicValue(a, A))), B)轉換成物理節點(FilterExec)，根據上面兩種場景會生成兩種不同的FilterExec

• b是分區列
  b是分區列，采集的是a列的所有值,如:

  case class DynamicPartitionPruneFilterExec(child: SparkPlan, collectors: Seq[(Expression, SparkPlan)])extends DynamicFilterUnaryExecNode with CodegenSupport with PredicateHelper

其中colletors就是用于采集信息的SparkPlan，因為要跑一個SQL來采集a列的所有值(select a from A group by a);
因為有可能會有多個分區列，所以這個地方是一個Seq.

• b是非分區列
  b是非分區列，采集的是a列的min/max/bloomfilter統計信息，如

  case class DynamicMinMaxFilterExec(child: SparkPlan, collectors: Seq[(Expression, SparkPlan)]) extends DynamicFilterUnaryExecNode with CodegenSupport with PredicateHelper

同理上面collectors也是用戶采集信息的SparkPlan，如select min(a),max(a) from A

執行

在物理執行計劃實際執行的過程中，會在DynamicPartitionPruneFilterExec/DynamicMinMaxFilterExec物理算子內先執行collectors獲取到a列的相關信息，然后對底層B的執行計劃進行改寫，比如利用采集到的信息做分區裁剪/數據過濾等。

Runtime Filter性能測試

以TPC-DS 10TB的Query54為例:

Runtime Filter 關閉

??

Runtime Filter 打開

經過DynamicPartitionPruneFilter對catalog_sales的分區進行了裁剪，實際對表的掃描從14,327,953,968減少到136,107,053，然后經過min/max的過濾繼續減少到135,564,763;另外Runtime Filter減少了大表的掃描，shuffle的數據量以及參加Join的數據量，所以對整個集群IO/網絡/CPU有比較大的節省

總結

針對Equi-Join的場景，可以額外的采集小表側的信息，然后在Join之前對大表進行分區裁剪或者掃描后過濾，從而提高查詢性能，減少資源消耗。

原文鏈接
本文為云棲社區原創內容，未經允許不得轉載。

總結

以上是生活随笔為你收集整理的EMR Spark Runtime Filter性能优化的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。