A sample code for QueryExectionListener from Spark 3.0.0

query_execution_listener_example.scala

// Register listener
spark
  .listenerManager
  .register(new QueryExecutionListener {
    override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = {
      val num = qe.observedMetrics
        .get("my_metrics")
        .map(_.getAs[Long]("num"))
        .getOrElse(-100.0)

      println(s"num of data: $num")
    }

    override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit = {}
  })

// Make DataFrame
val df = Seq
  .range(0, 1000)
  .map((_, Seq("a", "b", "c")(Random.nextInt(3)), math.random()))
  .toDF("id", "type", "value")

// Observe and process
val dfResult = df
  .observe("my_metrics", count($"*").as("num"))
  .groupBy($"type")
  .agg(avg($"value").as("avg_value"))

// Run
dfResult.show