Spark montre les statistiques de l'optimiseur basé sur les coûts

Question

J'ai essayé d'activer le cbo Spark en fixant la propriété dans spark-shell spark.conf.set("spark.sql.cbo.enabled", true)

Je suis en train d'exécuter spark.sql("ANALYZE TABLE events COMPUTE STATISTICS").show

L'exécution de cette requête ne me montre pas de statistiques spark.sql("select * from events where eventID=1").explain(true)

Exécution sur 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

Pour moi, les stats ne sont pas visibles en df.explain(true) . J'ai joué un peu avec, et j'ai pu imprimer des statistiques en utilisant println(df.queryExecution.stringWithStats) , exemple complet :

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)

donne

== 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)

Ceci n'apparaît pas dans la norme df.explain parce que cela se déclenche (Dataset.scala) :

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

Pour permettre l'édition des coûts, nous pouvons invoquer ce qui suit ExplainCommand nous-mêmes :

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))
}

Ici, vous pouvez également activer la sortie du code généré (set codegen = true )

Alternativement, ceci donne un résultat similaire

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

Pour voir les statistiques dans SparkUI, vous devez aller dans l'onglet SQL, puis sélectionner la requête correspondante (dans ce cas-ci df.show() ) :