Spark Show статистика на основе затрат оптимизатора

Question

Я попытался включить Spark cbo, установив свойство в spark-shell spark.conf.set("spark.sql.cbo.enabled", true)

Я сейчас работаю spark.sql("ANALYZE TABLE events COMPUTE STATISTICS").show

Выполнение этого запроса не показывает мне ничегостатистика spark.sql("select * from events where eventID=1").explain(true)

Запуск этого на Spark 2.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))
}

Здесь вы также можете включить вывод сгенерированногоcode (set codegen = true)

альтернативно, это дает аналогичный вывод

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

Чтобы просмотреть статистику в SparkUI, необходимо перейти на вкладку SQL, а затем выбрать соответствующий запрос (вэтот случай df.show()):