lucacanali / sparkmeasure Goto Github PK

It is not possible to call:

  def sendReportPrometheus(serverIPnPort: String,
                 metricsJob: String,
                 labelName: String = sparkSession.sparkContext.appName,
                 labelValue: String = sparkSession.sparkContext.applicationId): Unit

From Python.

Hi there,

thanks for all the resources you've provided to get a better understanding of spark metrics. Especially, I found your blog post very useful.

When using sparkMeasure, I still found it hard to come up with a proper interpretation for all the resulting task metrics. There are some easy once like CPU utilization (as described in your blog post). However, many others do not seem so trivial. For example:

• How to interpret sum(memoryBytesSpilled) and sum(diskBytesSpilled)? From my understanding, spilling occurs if data size exceeds available memory of an executor. memoryBytesSpilled refers to the deserialized size in memory whereas diskBytesSpilled refers to serialized size on disk for the data that is required to be spilled (see here and here). Temporarily storing data on disk is a performance penalty and should be avoided.
• How to interpret shuffleBytesWritten? Does this encompass the entire data which was required to be shuffled between all executors?
• How to interpret bytesread and byteswritten? Does this relate to the total amount of bytes read as input and bytes written as output (also from multiple sources/targets like HDFS, Kudu, Cassandra within the same job)?

What I'm basically aiming at is a beginners tutorial to guide new users through all the metrics with an example of what they mean and how they might be relevant for performance (e.g. spilling = bad).

For now, it would be great if you could help me better understand the 3 questions raised above. In future terms, it would be awesome to start a small guide on how to interpret spark metrics for the entire spark community (I would be in for that). Perhaps there is already an existing one but I couldn't find anything appropriate neither in the official docs nor on some personal projects.

Good day sir. I am a graphic designer and i am interested in designing a logo for your good project. I will be doing it as a gift for free. I just need your permission first before I begin my design. Hoping for your positive feedback. Thanks

This would provide an easy way for SparkMeasure users to fan-off all the events to an HTTP API endpoint (OpenLineage is another spark listener that provides a similar HTTP sink support)

Execuse me, I'm having some problem when getting the execution time of each operator in Spark SQL. For example, I'm using the belowed sql to test:
"select count(*) from range(1000) cross join range(1000) cross join range(1000)"
and I found some details in the Spark Web UI like this:

I wish to get the execution time of every operator, but operators like "RANGE, PROJECT etc" doesn't have the corresponding metrics.
I look into the log and I find that those operator doesn't have any metrics, so maybe this is the reason.
But I want to get execution time for each operator, how can I do that?

Trying to use sparkMeasure in scala spark application.
Get this error:

java.lang.NoSuchMethodError: org.apache.spark.scheduler.SparkListenerExecutorMetricsUpdate.executorUpdates()Lscala/collection/Map;```
scalaVersion:2.12.13
sparkVersion: 2.4.6.0
sparkMeasure: 0.22

Hi,

I am evaluating sparkMeasure for my use case but I always gets jobID filed as "0" or a series of 1. Is this expected behavior or I am missing something here?

Below are the outputs of a spark2-submit pi job with sparkMeasure and from a spark-shell job resp.

+-----+--------+-------+--------------------+
|jobId|jobGroup|stageId| name|
+-----+--------+-------+--------------------+
| 0| null| 0|reduce at SparkPi...|
+-----+--------+-------+--------------------+

+-----+--------+
|jobId|jobGroup|
+-----+--------+
| 0| null|
| 1| null|
| 1| null|
| 1| null|
| 1| null|
| 1| null|
| 1| null|

Also, how can we figure out to which task a particular metric belongs to?

Thanks
Amit

TasksMetrics and StageMetrics classes does not extend a common trait and this can cause some trouble while implementing those metrics in a generic way.

An example of a code that does not compile because of that:

val someExternalConfiguration = ExternalConfiguration.read();
val dataframe = spark.sql("SELECT * FROM SOME_WHERE")

someExternalConfiguration match {
    case "stages" => 
        val stagesMetrics = ch.cern.sparkmeasure.StageMetrics(spark)
        Writer.doWrite(dataFrame, stagesMetrics)
    case "tasks" => 
        val tasksMetrics = ch.cern.sparkmeasure.TaskMetrics(spark)
        Writer.doWrite(dataFrame, tasksMetrics)
}

object Writer {
    def doWrite(dataFrame: DataFrame, metrics: <Here should be the common trait>) {
        metrics.runAndMeasure(dataFrame.write.format("parquet").save("/tmp/any_where"))
    }
}

I am seeing an occasional exception when using the .report method on a TaskMetrics to render the metrics data as a string.

The problem seems to be associated with this line:
https://github.com/LucaCanali/sparkMeasure/blob/master/src/main/scala/ch/cern/sparkmeasure/taskmetrics.scala#L206

Here is the top of the stacktrace.

18/12/17 20:53:14 ERROR Uncaught throwable from user code: scala.MatchError: (elapsedTime,null) (of class scala.Tuple2)
	at ch.cern.sparkmeasure.TaskMetrics$$anonfun$report$1.apply(taskmetrics.scala:206)
	at ch.cern.sparkmeasure.TaskMetrics$$anonfun$report$1.apply(taskmetrics.scala:206)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
	at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
	at ch.cern.sparkmeasure.TaskMetrics.report(taskmetrics.scala:206)
	at com.ionic.helperfunctions.SparkMeasureHelpers$.save(SparkMeasure.scala:116)

It looks like this is probably likely caused by calling .report when there are no records in listenerTask.taskMetricsData (I'll try to confirm this).

If this is the case, there are a few options for fixing. The most obvious one is to change the match statement to something like:

      .map {
        case ((n: String, v: Long)) => Utils.prettyPrintValues(n, v)
        case ((n: String, null)) => n + " => null"
      }
     ).mkString("\n")

But there are, of course, other options.

File "C:\Users\talathar\Miniconda3\envs\XXXXX\lib\site-packages\sparkmeasure\stagemetrics.py", line 15, in init
self.stagemetrics = self.sc._jvm.ch.cern.sparkmeasure.StageMetrics(self.sparksession._jsparkSession)
File "C:\spark-3.2.1-bin-hadoop3.2\python\lib\py4j-0.10.9.3-src.zip\py4j\java_gateway.py", line 1586, in call
File "C:\spark-3.2.1-bin-hadoop3.2\python\pyspark\sql\utils.py", line 111, in deco
return f(*a, **kw)
File "C:\spark-3.2.1-bin-hadoop3.2\python\lib\py4j-0.10.9.3-src.zip\py4j\protocol.py", line 328, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling None.ch.cern.sparkmeasure.StageMetrics.
: java.lang.NoClassDefFoundError: scala/Product$class

I think this would be easier for users to check the serialized metrics if it was saved in a text format like JSON. I understand that the processing is done by code but it useful if the formats are human readable.

I get following error as I try to import sparkMeasure to my project:

[SUCCESSFUL ] ch.cern.sparkmeasure#spark-measure_2.12;0.22!spark-measure_2.12.jar (190ms)
[warn] 	::::::::::::::::::::::::::::::::::::::::::::::
[warn] 	::          UNRESOLVED DEPENDENCIES         ::
[warn] 	::::::::::::::::::::::::::::::::::::::::::::::
[warn] 	:: com.squareup.retrofit2#retrofit;2.4.0: Resolution failed several times for dependency: com.squareup.retrofit2#retrofit;2.4.0 {compile=[compile(*), master(*)], runtime=[runtime(*)]}::
[warn] 	com.squareup.retrofit2#retrofit;2.4.0!retrofit.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit/2.4.0/retrofit-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit;2.4.0!retrofit.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit/2.4.0/retrofit-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit;2.4.0!retrofit.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit/2.4.0/retrofit-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit;2.4.0!retrofit.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit/2.4.0/retrofit-2.4.0.pom
[warn] 	:: com.squareup.retrofit2#converter-moshi;2.4.0: Resolution failed several times for dependency: com.squareup.retrofit2#converter-moshi;2.4.0 {compile=[compile(*), master(*)], runtime=[runtime(*)]}::
[warn] 	Resolution failed several times for dependency: com.squareup.retrofit2#retrofit-converters;2.4.0 {}::
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn]
[warn] 	Resolution failed several times for dependency: com.squareup.retrofit2#retrofit-converters;2.4.0 {}::
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn]
[warn] 	Resolution failed several times for dependency: com.squareup.retrofit2#retrofit-converters;2.4.0 {}::
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn]
[warn] 	Resolution failed several times for dependency: com.squareup.retrofit2#retrofit-converters;2.4.0 {}::
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn] 	com.squareup.retrofit2#retrofit-converters;2.4.0!retrofit-converters.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/retrofit2/retrofit-converters/2.4.0/retrofit-converters-2.4.0.pom
[warn]
[warn] 	:: com.squareup.okhttp3#okhttp;3.11.0: Resolution failed several times for dependency: com.squareup.okhttp3#okhttp;3.11.0 {compile=[compile(*), master(*)], runtime=[runtime(*)]}::
[warn] 	com.squareup.okhttp3#okhttp;3.11.0!okhttp.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/okhttp3/okhttp/3.11.0/okhttp-3.11.0.pom
[warn] 	com.squareup.okhttp3#okhttp;3.11.0!okhttp.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/okhttp3/okhttp/3.11.0/okhttp-3.11.0.pom
[warn] 	com.squareup.okhttp3#okhttp;3.11.0!okhttp.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/okhttp3/okhttp/3.11.0/okhttp-3.11.0.pom
[warn] 	com.squareup.okhttp3#okhttp;3.11.0!okhttp.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/okhttp3/okhttp/3.11.0/okhttp-3.11.0.pom
[warn] 	:: com.squareup.okhttp3#logging-interceptor;3.11.0: Resolution failed several times for dependency: com.squareup.okhttp3#logging-interceptor;3.11.0 {compile=[compile(*), master(*)], runtime=[runtime(*)]}::
[warn] 	com.squareup.okhttp3#logging-interceptor;3.11.0!logging-interceptor.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/okhttp3/logging-interceptor/3.11.0/logging-interceptor-3.11.0.pom
[warn] 	com.squareup.okhttp3#logging-interceptor;3.11.0!logging-interceptor.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/okhttp3/logging-interceptor/3.11.0/logging-interceptor-3.11.0.pom
[warn] 	com.squareup.okhttp3#logging-interceptor;3.11.0!logging-interceptor.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/okhttp3/logging-interceptor/3.11.0/logging-interceptor-3.11.0.pom
[warn] 	com.squareup.okhttp3#logging-interceptor;3.11.0!logging-interceptor.pom(pom.original) origin location must be absolute: file:/Users/anirudh.vyas/.m2/repository/com/squareup/okhttp3/logging-interceptor/3.11.0/logging-interceptor-3.11.0.pom
[warn] 	::::::::::::::::::::::::::::::::::::::::::::::
[warn]
[warn] 	Note: Unresolved dependencies path:
[warn] 		com.squareup.retrofit2:retrofit:2.4.0
[warn] 		  +- org.influxdb:influxdb-java:2.14
[warn] 		  +- ch.cern.sparkmeasure:spark-measure_2.12:0.22 (Defaults.scala#L4318)
[warn] 		  +- com.workday.perftool:spark-commons_2.12:0.0.34-SNAPSHOT
[warn] 		com.squareup.retrofit2:converter-moshi:2.4.0
[warn] 		  +- org.influxdb:influxdb-java:2.14
[warn] 		  +- ch.cern.sparkmeasure:spark-measure_2.12:0.22 (Defaults.scala#L4318)
[warn] 		  +- com.workday.perftool:spark-commons_2.12:0.0.34-SNAPSHOT
[warn] 		com.squareup.okhttp3:okhttp:3.11.0
[warn] 		  +- org.influxdb:influxdb-java:2.14
[warn] 		  +- ch.cern.sparkmeasure:spark-measure_2.12:0.22 (Defaults.scala#L4318)
[warn] 		  +- com.workday.perftool:spark-commons_2.12:0.0.34-SNAPSHOT
[warn] 		com.squareup.okhttp3:logging-interceptor:3.11.0
[warn] 		  +- org.influxdb:influxdb-java:2.14
[warn] 		  +- ch.cern.sparkmeasure:spark-measure_2.12:0.22 (Defaults.scala#L4318)
[warn] 		  +- com.workday.perftool:spark-commons_2.12:0.0.34-SNAPSHOT

What should I do differently?

are there any barriers to using this with a java spark job?

Scala classes are java classes, so I think it should be fine: https://stackoverflow.com/questions/44993014/can-scala-classes-be-used-in-java#:~:text=Scala%20classes%20are%20Java%20classes,in%20Java%2C%20for%20example%20traits.

I'll report back

Is there support for InfluxDB 2.0 using Flight Recorder mode? The current credentials seem to only support username/password
authentication versus token/org credentials needed for InfluxDB 2.0

 val url = Utils.parseInfluxDBURL(conf, logger)
  val (username, password) = Utils.parseInfluxDBCredentials(conf, logger)

  // Tries to connect to InfluxDB, using the given URL and credentials
  val influxDB =  username match {
    case username if username.isEmpty =>
      // no username and password, InfluxDB must be running with auth-enabled=false
      InfluxDBFactory.connect(url)
    case _ => InfluxDBFactory.connect(url, username, password)
  }

I got an error after run following command

val stageMetrics = ch.cern.sparkmeasure.StageMetrics(spark) 

stageMetrics.begin()

spark.sql("SELECT 1=1")

stageMetrics.end()
stageMetrics.printReport()

got following error

at ch.cern.sparkmeasure.StageMetrics$$anonfun$report$1.apply(stagemetrics.scala:184)
	at ch.cern.sparkmeasure.StageMetrics$$anonfun$report$1.apply(stagemetrics.scala:184)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
	at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
	at ch.cern.sparkmeasure.StageMetrics.report(stagemetrics.scala:184)
	at ch.cern.sparkmeasure.StageMetrics.printReport(stagemetrics.scala:193)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read$$iw$$iw$$iw$$iw$$iw$$iw.<init>(command-652119476452024:3)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read$$iw$$iw$$iw$$iw$$iw.<init>(command-652119476452024:48)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read$$iw$$iw$$iw$$iw.<init>(command-652119476452024:50)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read$$iw$$iw$$iw.<init>(command-652119476452024:52)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read$$iw$$iw.<init>(command-652119476452024:54)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read$$iw.<init>(command-652119476452024:56)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read.<init>(command-652119476452024:58)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read$.<init>(command-652119476452024:62)
	at linea8b963cf2ca141a2ac1e991add9914c629.$read$.<clinit>(command-652119476452024)
	at linea8b963cf2ca141a2ac1e991add9914c629.$eval$.$print$lzycompute(<notebook>:7)
	at linea8b963cf2ca141a2ac1e991add9914c629.$eval$.$print(<notebook>:6)

SLF4J is a nice logging abstraction that allows developers include logging but lets users choose their preferred logging framework.
Could you switch to having a compile dependency on slf4j-api jar and a test dependency on log4j and the log4j-slf4j-impl?

I am wondering if you still have a sample of jupyter notebook with sparkmeasure
The following SparkMeasure_Jupyer_Python_getting_started.ipynb is not found as of today
Thanks in advance

Issue

When trying to use the Flight Recorder functionality, I get the following error:

java.lang.NoSuchMethodError: com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper.$init$(Lcom/fasterxml/jackson/module/scala/experimental/ScalaObjectMapper;)V
        at ch.cern.sparkmeasure.IOUtils$$anon$1.<init>(ioutils.scala:22)
        at ch.cern.sparkmeasure.IOUtils$.<init>(ioutils.scala:22)
        at ch.cern.sparkmeasure.IOUtils$.<clinit>(ioutils.scala)
        at ch.cern.sparkmeasure.FlightRecorderStageMetrics.onApplicationEnd(flightrecorder.scala:56)
        at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:57)
        at org.apache.spark.scheduler.SparkListenerBus.doPostEvent$(SparkListenerBus.scala:28)
        at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
        at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
        at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:117)
        at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:101)
        at org.apache.spark.scheduler.AsyncEventQueue.super$postToAll(AsyncEventQueue.scala:105)
        at org.apache.spark.scheduler.AsyncEventQueue.$anonfun$dispatch$1(AsyncEventQueue.scala:105)
        at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
        at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
        at org.apache.spark.scheduler.AsyncEventQueue.org$apache$spark$scheduler$AsyncEventQueue$$dispatch(AsyncEventQueue.scala:100)
        at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.$anonfun$run$1(AsyncEventQueue.scala:96)
        at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1404)
        at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.run(AsyncEventQueue.scala:96)

Reproduce

To reproduce, checkout the latest version of the code and run:

spark-submit --master local[*] --jars ./target/scala-2.12/spark-measure_2.12-0.18-SNAPSHOT.jar --conf spark.extraListeners=ch.cern.sparkmeasure.FlightRecorderStageMetrics ~/pyspark_helloworld.py

where pyspark_helloworld.py is just:

import pyspark
sc = pyspark.SparkContext('local[*]')

txt = sc.textFile('file:////usr/share/doc/python-2.7.5/README')
print(txt.count())

python_lines = txt.filter(lambda line: 'python' in line.lower())
print(python_lines.count())

I am using the following versions:

• Scala 2.12
• PySpark 3.2.0
• Python 3.7.11

Fix

Bumping the version of jackson-module-scala to 2.13.0 and rebuilding seems to fix the problem.

Hi. I'm trying use this library.
Samples of code:

build.sbt

val hdpMinorVersion = "3.1.0.0-78"
val hadoopVersion = "3.1.1" + "." + hdpMinorVersion
val sparkVersion = "2.3.2" + "." + hdpMinorVersion


lazy val localResolvers = Seq(
  "mvnrepository" at "https://mvnrepository.com/artifact/",
  "Hortonworks HDP" at "http://repo.hortonworks.com/content/repositories/releases/",
  "Hortonworks Other Dependencies" at "http://repo.hortonworks.com/content/groups/public"
)

val projectResolvers: Seq[Resolver] = Seq(Resolver.defaultLocal, Resolver.mavenLocal) ++ localResolvers

resolvers := projectResolvers

lazy val sparkDependencies = Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion % Provided,
  "org.apache.spark" %% "spark-sql" % sparkVersion % Provided
)

lazy val hdpDependencies = Seq(
  "com.hortonworks.hive" %% "hive-warehouse-connector" % "1.0.0.3.0.1.0-187" % Provided intransitive()
)

lazy val staticAnalyzer = Seq(
  compilerPlugin(dependency = "org.wartremover" %% "wartremover" % "2.3.4")
)

libraryDependencies ++= sparkDependencies ++ hdpDependencies ++ staticAnalyzer ++ Seq(
  "io.monix" %% "monix" % "2.3.3",
  "org.typelevel" %% "cats-core" % "0.9.0",
  //  "io.monix" %% "monix-eval" % "2.3.3",
  "ch.cern.sparkmeasure" %% "spark-measure" % "0.13",
  "io.monix" %% "monix-cats" % "2.3.3",
  "org.scalatest" %% "scalatest" % "2.2.6" % "test"
)

spark-shell

scala> val taskMetrics = ch.cern.sparkmeasure.TaskMetrics(spark)
taskMetrics: ch.cern.sparkmeasure.TaskMetrics = TaskMetrics(org.apache.spark.sql.SparkSession@69c0bae6,false)

scala> taskMetrics.runAndMeasure
   def runAndMeasure[T](f: => T): T

scala> taskMetrics.runAndMeasure(spark.sql("select * from test.test"))
Hive Session ID = 04e7280b-0a45-4fad-867f-f1447faf6bf4
Time taken: 4895 ms
19/02/20 09:01:58 WARN Utils: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.
19/02/20 09:01:58 WARN TaskMetrics: Stage metrics data refreshed into temp view PerfTaskMetrics
scala.MatchError: (elapsedTime,null) (of class scala.Tuple2)                    
  at ch.cern.sparkmeasure.TaskMetrics$$anonfun$report$1.apply(taskmetrics.scala:206)
  at ch.cern.sparkmeasure.TaskMetrics$$anonfun$report$1.apply(taskmetrics.scala:206)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
  at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
  at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
  at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
  at ch.cern.sparkmeasure.TaskMetrics.report(taskmetrics.scala:206)
  at ch.cern.sparkmeasure.TaskMetrics.printReport(taskmetrics.scala:215)
  at ch.cern.sparkmeasure.TaskMetrics.runAndMeasure(taskmetrics.scala:282)
  ... 49 elided

What I'm doing wrong?

I'm trying to use the deserializer. However the toDF that follows is throwing an exception:

Details:
val taskVals = ch.cern.sparkmeasure.Utils.readSerializedTaskMetrics("/tmp/stageMetrics.serialized")
scala> val taskVals = ch.cern.sparkmeasure.Utils.readSerializedTaskMetrics("/tmp/stageMetrics.serialized")
taskVals: scala.collection.mutable.ListBuffer[ch.cern.sparkmeasure.TaskVals] = ListBuffer(StageVals(0,0,parquet at NativeMethodAccessorImpl.java:0,1531905390534,1531905392912,2378,1,1878,123,368,259,5,78,4802,2,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0), StageVals(1,1,count at NativeMethodAccessorImpl.java:0,1531905395225,1531905405201,9976,1003,1028843,104666,103972,21334,218,35221,1714153,144,0,0,0,100818815,11697322,0,0,0,0,0,0,0,1123,56844,1003), StageVals(1,2,count at NativeMethodAccessorImpl.java:0,1531905405221,1531905405591,370,1,336,176,26,19,1,0,2382,2,0,0,0,0,0,0,0,55,56844,1003,28,975,0,0,0), StageVals(2,3,collect at /root/SparCle/workload/sqlquery/data-layout-read.py:20,1531905405925,1531905413411,7486,1003,813482,25942,6350,1765,58,9238,1707192,144,0,0,0,100818815,156304569,0,0,0,0,0,...
scala> val taskMetricsDF = taskVals.toDF()
java.lang.RuntimeException: Error while encoding: java.lang.ClassCastException: ch.cern.sparkmeasure.StageVals incompatible with ch.cern.sparkmeasure.TaskVals
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).jobId AS jobId#444
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).stageId AS stageId#445
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).index AS index#446L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).launchTime AS launchTime#447L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).finishTime AS finishTime#448L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).duration AS duration#449L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).schedulerDelay AS schedulerDelay#450L
staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).executorId, true) AS executorId#451
staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).host, true) AS host#452
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).taskLocality AS taskLocality#453
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).speculative AS speculative#454
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).gettingResultTime AS gettingResultTime#455L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).successful AS successful#456
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).executorRunTime AS executorRunTime#457L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).executorCpuTime AS executorCpuTime#458L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).executorDeserializeTime AS executorDeserializeTime#459L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).executorDeserializeCpuTime AS executorDeserializeCpuTime#460L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).resultSerializationTime AS resultSerializationTime#461L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).jvmGCTime AS jvmGCTime#462L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).resultSize AS resultSize#463L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).numUpdatedBlockStatuses AS numUpdatedBlockStatuses#464
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).diskBytesSpilled AS diskBytesSpilled#465L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).memoryBytesSpilled AS memoryBytesSpilled#466L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).peakExecutionMemory AS peakExecutionMemory#467L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).recordsRead AS recordsRead#468L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).bytesRead AS bytesRead#469L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).recordsWritten AS recordsWritten#470L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).bytesWritten AS bytesWritten#471L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).shuffleFetchWaitTime AS shuffleFetchWaitTime#472L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).shuffleTotalBytesRead AS shuffleTotalBytesRead#473L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).shuffleTotalBlocksFetched AS shuffleTotalBlocksFetched#474L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).shuffleLocalBlocksFetched AS shuffleLocalBlocksFetched#475L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).shuffleRemoteBlocksFetched AS shuffleRemoteBlocksFetched#476L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).shuffleWriteTime AS shuffleWriteTime#477L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).shuffleBytesWritten AS shuffleBytesWritten#478L
assertnotnull(assertnotnull(input[0, ch.cern.sparkmeasure.TaskVals, true])).shuffleRecordsWritten AS shuffleRecordsWritten#479L
at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:290)
at org.apache.spark.sql.SparkSession$$anonfun$2.apply(SparkSession.scala:464)
at org.apache.spark.sql.SparkSession$$anonfun$2.apply(SparkSession.scala:464)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.List.foreach(List.scala:381)
at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35)
at scala.collection.mutable.ListBuffer.foreach(ListBuffer.scala:45)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:464)
at org.apache.spark.sql.SQLContext.createDataset(SQLContext.scala:377)
at org.apache.spark.sql.SQLImplicits.localSeqToDatasetHolder(SQLImplicits.scala:213)
... 48 elided
Caused by: java.lang.ClassCastException: ch.cern.sparkmeasure.StageVals incompatible with ch.cern.sparkmeasure.TaskVals
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply_0$(Unknown Source)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:287)
... 60 more

scala>

@LucaCanali would it be possible for you to login to Travis CI with your github credentials and enable Travis builds for sparkMeasure?

You could then add these badges at the top of your README.md.
These are copied from https://github.com/swagger-akka-http/swagger-akka-http/blob/master/README.md

[![Build Status](https://travis-ci.org/LucaCanali/sparkMeasure.svg?branch=master)](https://travis-ci.org/LucaCanali/sparkMeasure)
[![Maven Central](https://maven-badges.herokuapp.com/maven-central/LucaCanali/sparkMeasure/badge.svg)](https://maven-badges.herokuapp.com/maven-central/LucaCanali/sparkMeasure)

Hi,

While trying to run the following example code:
https://github.com/LucaCanali/sparkMeasure/tree/master/examples/testSparkMeasureScala

I get the following error when running sbt package.

What am I doing wrong?

C:\repos\sparkMeasure\examples\testSparkMeasureScala> sbt package [info] welcome to sbt 1.7.1 (Oracle Corporation Java 19.0.1) [info] loading global plugins from C:\Users\DanielAronovich\.sbt\1.0\plugins [info] loading project definition from C:\repos\sparkMeasure\examples\testSparkMeasureScala\project [info] loading settings for project testsparkmeasurescala from build.sbt ... [info] set current project to testSparkMeasureScala (in build file:/C:/repos/sparkMeasure/examples/testSparkMeasureScala/) [warn] [warn] Note: Unresolved dependencies path: [error] sbt.librarymanagement.ResolveException: Error downloading ch.cern.sparkmeasure:spark-measure_2.12:0.21 [error] Not found [error] Not found [error] not found: C:\Users\DanielAronovich\.ivy2\localch.cern.sparkmeasure\spark-measure_2.12\0.21\ivys\ivy.xml [error] not found: https://repo1.maven.org/maven2/ch/cern/sparkmeasure/spark-measure_2.12/0.21/spark-measure_2.12-0.21.pom [error] at lmcoursier.CoursierDependencyResolution.unresolvedWarningOrThrow(CoursierDependencyResolution.scala:345) [error] at lmcoursier.CoursierDependencyResolution.$anonfun$update$38(CoursierDependencyResolution.scala:314) [error] at scala.util.Either$LeftProjection.map(Either.scala:573) [error] at lmcoursier.CoursierDependencyResolution.update(CoursierDependencyResolution.scala:314) [error] at sbt.librarymanagement.DependencyResolution.update(DependencyResolution.scala:60) [error] at sbt.internal.LibraryManagement$.resolve$1(LibraryManagement.scala:59) [error] at sbt.internal.LibraryManagement$.$anonfun$cachedUpdate$12(LibraryManagement.scala:133) [error] at sbt.util.Tracked$.$anonfun$lastOutput$1(Tracked.scala:73) [error] at sbt.internal.LibraryManagement$.$anonfun$cachedUpdate$20(LibraryManagement.scala:146) [error] at scala.util.control.Exception$Catch.apply(Exception.scala:228) [error] at sbt.internal.LibraryManagement$.$anonfun$cachedUpdate$11(LibraryManagement.scala:146) [error] at sbt.internal.LibraryManagement$.$anonfun$cachedUpdate$11$adapted(LibraryManagement.scala:127) [error] at sbt.util.Tracked$.$anonfun$inputChangedW$1(Tracked.scala:219) [error] at sbt.internal.LibraryManagement$.cachedUpdate(LibraryManagement.scala:160) [error] at sbt.Classpaths$.$anonfun$updateTask0$1(Defaults.scala:3688) [error] at scala.Function1.$anonfun$compose$1(Function1.scala:49) [error] at sbt.internal.util.$tilde$greater.$anonfun$$u2219$1(TypeFunctions.scala:62) [error] at sbt.std.Transform$$anon$4.work(Transform.scala:68) [error] at sbt.Execute.$anonfun$submit$2(Execute.scala:282) [error] at sbt.internal.util.ErrorHandling$.wideConvert(ErrorHandling.scala:23) [error] at sbt.Execute.work(Execute.scala:291) [error] at sbt.Execute.$anonfun$submit$1(Execute.scala:282) [error] at sbt.ConcurrentRestrictions$$anon$4.$anonfun$submitValid$1(ConcurrentRestrictions.scala:265) [error] at sbt.CompletionService$$anon$2.call(CompletionService.scala:64) [error] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317) [error] at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:577) [error] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317) [error] at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) [error] at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) [error] at java.base/java.lang.Thread.run(Thread.java:1589) [error] (update) sbt.librarymanagement.ResolveException: Error downloading ch.cern.sparkmeasure:spark-measure_2.12:0.21 [error] Not found [error] Not found [error] not found: C:\Users\DanielAronovich\.ivy2\localch.cern.sparkmeasure\spark-measure_2.12\0.21\ivys\ivy.xml [error] not found: https://repo1.maven.org/maven2/ch/cern/sparkmeasure/spark-measure_2.12/0.21/spark-measure_2.12-0.21.pom [error] Total time: 4 s, completed Nov 15, 2022, 3:11:11 PM

Hello,
I am trying the Google Colab example. I get

TypeError: 'JavaPackage' object is not callable

when running:
stagemetrics = StageMetrics(spark)

Are there any other settings to do with Google Colab?
Thank you!

We are using S3 compatible Object Storage for Spark Storage but the current default IO metrics support hdfs only.

SELECT non_negative_derivative("value", 1s) FROM "filesystem.hdfs.read_bytes" WHERE "applicationid" = '$ApplicationId' AND $timeFilter GROUP BY process

Is there anyway can able to fetch other distribute file system IO metrics ?

Thanks !

I launched sparkMeasure in a large job. Immediately I got:
18/02/07 08:21:56 ERROR org.apache.spark.scheduler.LiveListenerBus: Dropping SparkListenerEvent because no remaining room in event queue. This likely means one of the SparkListeners is too slow and cannot keep up with the rate at which tasks are being started by the scheduler.
18/02/07 08:21:56 WARN org.apache.spark.scheduler.LiveListenerBus: Dropped 1 SparkListenerEvents since Thu Jan 01 00:00:00 UTC 1970
18/02/07 08:22:56 WARN org.apache.spark.scheduler.LiveListenerBus: Dropped 13971 SparkListenerEvents since Wed Feb 07 08:21:56 UTC 2018
18/02/07 08:23:51 ERROR org.apache.spark.network.server.TransportRequestHandler: Error opening block StreamChunkId{streamId=1999850815777, chunkIndex=0} for request from /10.205.151.192:37514 org.apache.spark.storage.BlockNotFoundException: Block broadcast_32_piece0 not found

The job does continue, but it seems to be overloading the listenerbus.
I'll try --conf spark.scheduler.listenerbus.eventqueue.size=100000.

Did you already encounter this somewhere?

self.stagemetrics = self.sc._jvm.ch.cern.sparkmeasure.StageMetrics(self.sparksession._jsparkSession) TypeError: 'JavaPackage' object is not callable issue is hit when I spark-submit the .py script to emr.

Hi Luca and community,

first of all - thanks for the great work - very much appreciate it!

This is not a real issue but rather an user question asking for clarification. I've been wondering if sparkMeasure provides any additional metrics than the default REST API?

I would like to collect spark job metrics while keeping any dependencies as minimal as possible. Using the default REST API to collect the metrics seems simple without needing to rely on an additional package. Of course sparkMeasure provides additional abstractions to aggregate on stage/task level and to compute many relevant metrics. That is of great use. We are likely to be interested in only a few core metrics and we don't need all of them.

Are there any plans to support bindings for R to measure performance metrics via sparkR?

Found an odd issue. We recently started running our jobs through Azure Synapse. While using Azure HDI, we were able to record the peakExecutionMemory, but for some reason with Azure Synapse, all the values are 0.

We are using TaskMetrics to get the most information out of the run and within the csv generated, other columns are populated except for the peakExecutionMemory which are all 0.

Is this a known issue?

We are running with python 3.7, pyspark 3.2.1, and scala 2.12 and using the spark-measure_2.12:0.18.jar

Hi,

I am wondering if you have any workaround or recommendation for using flight recorder mode when the driver can crashed from OOM. When the driver crashed from OOM, the Listener would never received onApplicationEnd and no metrics would be written to the sink. Ideally, we would still want to know all the metrics of the jobs accumulated right before it crashed.

Can we get memory usage of a spark-submit? stagemetrics.print_report() don't have that.

when try to compile the testsparkmeasurescala_2.12

not found: https://repo1.maven.org/maven2/ch/cern/sparkmeasure/spark-measure_2.12/0.21/spark-measure_2.12-0.21.pom

Implement a prometheus sink in in Flight Recorder Mode that formats metrics into a Prometheus-friendly format and pushes them to a configured Prometheus Pushgateway endpoint.

sparkmeasure currently offers Prometheus sink capabilities, facilitating the monitoring of Spark performance metrics within Prometheus ecosystems. Extending this functionality to include support for Flight Recorder Mode represents a significant enhancement. This feature would enable users to push metrics directly to a Prometheus Pushgateway endpoint.

Hello!

I'm trying to create a dataframe to save my task metrics in Hive. However, when I call the create_taskmetrics_DF method, as below, the error occurs on the print screen.

tm = taskmetrics.create_taskmetrics_DF("PerfTaskMetrics")

It looks like an error in the class, looking only at the error message.

Thanks for this work @LucaCanali ! I had one question about I/O metrics, I know you have mentioned in the limitations sections of the README that Spark does not expose I/O and network related metrics. However I was wondering if there was any way to deduce approximately the time spent in I/O for a job given current metrics? For instance, what does the different between ExecutorRunTime and ExecutorCpuTime entail?

I used Spark to read local Parquet file and store it under different name.
I used file:// as i accessed file locally.

sum(bytesWritten) => 0 (0 Bytes) was reported 0. Why is this?

Thanks