Spark SQL where 子句中谓词的求值顺序

如何解决Spark SQL where 子句中谓词的求值顺序

我试图了解 Spark SQL 中谓词评估的顺序，以提高查询的性能。
假设我有以下查询

"select * from tbl where pred1 and pred2"

假设没有一个谓词符合下推过滤器的条件（为了简化）。还假设 pred1 在计算上比 pred2 复杂得多（假设正则表达式模式匹配与否定）。

有什么方法可以验证 spark 会先评估 pred2 pred1？
这是确定性的吗？
这是可控的吗？
有没有办法查看最终执行计划？

解决方法

一般

好问题。

通过测试场景并在无法找到合适的文档时进行推断得出的答案。由于网络上的各种说法无法备份，第二次尝试。

我认为这个问题不是关于 AQE Spark 3.x 方面的，而是关于大约说，作为 Spark 应用程序阶段 N 一部分的数据帧具有通过了从静止源获取数据的阶段，即受过滤并应用多个谓词。

那么中心点是谓词如何重要排序或 Spark (Catalyst) 重新排序谓词以最小化要完成的工作？

这里的前提是首先过滤掉最大数量的数据比评估一个过滤非常多的谓词更有意义很少出去。
- 这是一个众所周知的 RDBMS 点，指的是 sargable 谓词（取决于定义随时间的演变）。
  - 很多讨论都集中在索引上，Spark、Hive 没有这个，但 DF 是柱状的。

第 1 点

您可以尝试%sql

 EXPLAIN EXTENDED select k,sum(v) from values (1,2),(1,3) t(k,v) group by k;

从这里你可以看到如果重新安排了会发生什么谓词，但我在非 AQE 的物理计划中没有看到这些方面 Databricks 上的模式。参考 https://docs.databricks.com/sql/language-manual/sql-ref-syntax-qry-explain.html。

Catalyst 可以重新安排我在这里和那里阅读的过滤。要什么程度，是大量的研究；我无法确认这一点。

还有一个有趣的读物： https://www.waitingforcode.com/apache-spark-sql/catalyst-optimizer-in-spark-sql/read

第 2 点

我用相同的方式运行了以下可悲的人为示例函数式查询，但谓词颠倒，使用具有高基数并测试了实际上不存在的值然后在调用时比较 UDF 中使用的累加器的计数。

场景一

import org.apache.spark.sql.functions._

def randomInt1to1000000000 = scala.util.Random.nextInt(1000000000)+1
def randomInt1to10 = scala.util.Random.nextInt(10)+1
def randomInt1to1000000 = scala.util.Random.nextInt(1000000)+1

val df = sc.parallelize(Seq.fill(1000000){(randomInt1to1000000,randomInt1to1000000000,randomInt1to10)}).toDF("nuid","hc","lc").withColumn("text",lpad($"nuid",3,"0")).withColumn("literal",lit(1)) 

val accumulator = sc.longAccumulator("udf_call_count")

spark.udf.register("myUdf",(x: String) => {accumulator.add(1)
                                            x.length}
                  )  

accumulator.reset()
df.where("myUdf(text) = 3 and hc = -4").select(max($"text")).show(false)
println(s"Number of UDF calls ${accumulator.value}")

+---------+
|max(text)|
+---------+
|null     |
+---------+

Number of UDF calls 1000000

场景 2

import org.apache.spark.sql.functions._

def randomInt1to1000000000 = scala.util.Random.nextInt(1000000000)+1
def randomInt1to10 = scala.util.Random.nextInt(10)+1
def randomInt1to1000000 = scala.util.Random.nextInt(1000000)+1

val dfA = sc.parallelize(Seq.fill(1000000){(randomInt1to1000000,(x: String) => {accumulator.add(1)
                                            x.length}
                  )  

accumulator.reset()
dfA.where("hc = -4 and myUdf(text) = 3").select(max($"text")).show(false)
println(s"Number of UDF calls ${accumulator.value}")

+---------+
|max(text)|
+---------+
|null     |
+---------+

Number of UDF calls 0

我的结论是：

有从左到右的评估 - 在这种情况下 - 因为有 0 个对 udf 的调用，因为场景 2 的累加器值为 0，而场景 1 注册了 1M 次调用。
因此，ORACLE 和 DB2 可能对 Stage 1 谓词执行的谓词处理顺序不适用。

第 3 点

但是我从手册中注意到 https://docs.databricks.com/spark/latest/spark-sql/udf-scala.html 以下：

求值顺序和空检查

Spark SQL（包括 SQL 和 DataFrame 和 Dataset API）不保证子表达式的求值顺序。特别是，不一定要评估运算符或函数的输入从左到右或以任何其他固定顺序。例如，逻辑与和 OR 表达式没有从左到右的“短路” 语义。

因此，依赖副作用或顺序是危险的布尔表达式的计算，以及 WHERE 和 HAVING 的顺序子句，因为这样的表达式和子句可以在查询优化和规划。具体来说，如果 UDF 依赖于 SQL 中用于空值检查的短路语义，没有保证在调用 UDF 之前会进行空检查。为了例如，

spark.udf.register("strlen",(s: String) => s.length)
spark.sql("select s from test1 where s is not null and strlen(s) > 1") // no guarantee

这个 WHERE 子句不保证调用 strlen UDF 过滤掉空值后。

要执行正确的空检查，我们建议您执行以下任一操作以下内容：

使 UDF 本身能够感知空值并在 UDF 内部进行空值检查本身使用 IF 或 CASE WHEN 表达式进行空检查并调用条件分支中的 UDF。

spark.udf.register("strlen_nullsafe",(s: String) => if (s != null) s.length else -1)
spark.sql("select s from test1 where s is not null and strlen_nullsafe(s) > 1") // ok
spark.sql("select s from test1 where if(s is not null,strlen(s),null) > 1")   // ok

有点矛盾。

Spark SQL where 子句中谓词的求值顺序

如何解决Spark SQL where 子句中谓词的求值顺序

解决方法

相关推荐