基于火花显示成本的优化器统计

Question

我尝试通过在星体spark.conf.set("spark.sql.cbo.enabled", true)中设置属性来启用星火cbo。

我现在正在运行spark.sql("ANALYZE TABLE events COMPUTE STATISTICS").show

运行此查询不会显示任何统计信息spark.sql("select * from events where eventID=1").explain(true)

在Spark2.2.1上运行这个

scala> spark.sql("select * from events where eventID=1").explain()
== Physical Plan ==
*Project [buyDetails.capacity#923, buyDetails.clearingNumber#924, buyDetails.leavesQty#925L, buyDetails.liquidityCode#926, buyDetails.orderID#927, buyDetails.side#928, cancelQty#929L, capacity#930, clearingNumber#931, contraClearingNumber#932, desiredLeavesQty#933L, displayPrice#934, displayQty#935L, eventID#936, eventTimestamp#937L, exchange#938, executionCodes#939, fillID#940, handlingInstructions#941, initiator#942, leavesQty#943L, nbbPrice#944, nbbQty#945L, nboPrice#946, ... 29 more fields]
+- *Filter (isnotnull(eventID#936) && (cast(eventID#936 as int) = 1))
+- *FileScan parquet default.events[buyDetails.capacity#923,buyDetails.clearingNumber#924,buyDetails.leavesQty#925L,buyDetails.liquidityCode#926,buyDetails.orderID#927,buyDetails.side#928,cancelQty#929L,capacity#930,clearingNumber#931,contraClearingNumber#932,desiredLeavesQty#933L,displayPrice#934,displayQty#935L,eventID#936,eventTimestamp#937L,exchange#938,executionCodes#939,fillID#940,handlingInstructions#941,initiator#942,leavesQty#943L,nbbPrice#944,nbbQty#945L,nboPrice#946,... 29 more fields] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/home/asehgal/data/events], PartitionFilters: [], PushedFilters: [IsNotNull(eventID)], ReadSchema: struct<buyDetails.capacity:string,buyDetails.clearingNumber:string,buyDetails.leavesQty:bigint,bu...

Answer 1

对我来说，这些数据在df.explain(true)中是不可见的。我玩了一会儿，可以使用println(df.queryExecution.stringWithStats)打印统计数据，完整的例子如下：

val ss = SparkSession
  .builder()
  .master("local[*]")
  .appName("TestCBO")
  .config("spark.sql.cbo.enabled",true)
  .getOrCreate()

import ss.implicits._

val df1 = ss.range(10000L).toDF("i")
df1.write.mode("overwrite").saveAsTable("table1")

val df2 = ss.range(100000L).toDF("i")
df2.write.mode("overwrite").saveAsTable("table2")

ss.sql("ANALYZE TABLE table1 COMPUTE STATISTICS FOR COLUMNS i")
ss.sql("ANALYZE TABLE table2 COMPUTE STATISTICS FOR COLUMNS i")

val df = ss.table("table1").join(ss.table("table2"), "i")
  .where($"i" > 1000)

println(df.queryExecution.stringWithStats)

给出

== Optimized Logical Plan ==
Project [i#2554L], Statistics(sizeInBytes=147.2 KB, rowCount=9.42E+3, hints=none)
+- Join Inner, (i#2554L = i#2557L), Statistics(sizeInBytes=220.8 KB, rowCount=9.42E+3, hints=none)
   :- Filter (isnotnull(i#2554L) && (i#2554L > 1000)), Statistics(sizeInBytes=140.6 KB, rowCount=9.00E+3, hints=none)
   :  +- Relation[i#2554L] parquet, Statistics(sizeInBytes=156.3 KB, rowCount=1.00E+4, hints=none)
   +- Filter ((i#2557L > 1000) && isnotnull(i#2557L)), Statistics(sizeInBytes=1546.9 KB, rowCount=9.90E+4, hints=none)
      +- Relation[i#2557L] parquet, Statistics(sizeInBytes=1562.5 KB, rowCount=1.00E+5, hints=none)

标准df.explain中没有显示这一点，因为这会触发(Dataset.scala)：

ExplainCommand(queryExecution.logical, extended = true) //  cost = false in this constructor

为了支持成本的输出，我们可以自己调用这个ExplainCommand：

import org.apache.spark.sql.execution.command.ExplainCommand
val explain = ExplainCommand(df.queryExecution.logical, extended = true, cost = true)
ss.sessionState.executePlan(explain).executedPlan.executeCollect().foreach {
  r => println(r.getString(0))
}

在这里，还可以启用生成代码的输出(设置为codegen = true)。

或者，这提供了类似的输出。

df // join of two dataframes and filter
 .registerTempTable("tmp")
ss.sql("EXPLAIN COST select * from tmp").show(false)

要查看SparkUI中的统计信息，必须转到SQL-tab，然后选择相应的查询(在本例中为df.show())：

​

​