SparkMeasure - a performance tool for Apache Spark

SparkMeasure is a tool and a library designed to ease performance measurement and troubleshooting of Apache Spark jobs. It focuses on easing the collection and analysis of Spark metrics, making it a practical choice for both developers and data engineers. With sparkMeasure, users gain a deeper understanding of their Spark job performance, enabling faster and more reliable data processing workflows.

✨ Highlights

• Interactive Troubleshooting: Ideal for real-time analysis of Spark workloads in notebooks and spark-shell/pyspark environments.
• Development & CI/CD Integration: Facilitates testing, measuring, and comparing execution metrics of Spark jobs under various configurations or code changes.
• Batch Job Analysis: With Flight Recorder mode sparkMeasure transparently records batch job metrics for later analysis.
• Monitoring Capabilities: Integrates with external systems like Apache Kafka, Prometheus Push Gateway, Prometheus JMX Exporter, and InfluxDB for extensive monitoring.
• Educational Tool: Serves as a practical example of implementing Spark Listeners for the collection of detailed Spark task metrics.
• Language Compatibility: Fully supports Scala, Java, and Python, making it versatile for a wide range of Spark applications.

📚 Table of Contents

• SparkMeasure - a performance tool for Apache Spark
  • ✨ Highlights
  • 📚 Table of Contents
  • Links to related work on Spark Performance
• 🚀 Quick start
  • Examples of sparkMeasure on notebooks
  • Examples of sparkMeasure on the CLI
    • Python CLI
    • Scala CLI
  • Memory report
  • CLI example for Task Metrics:
• Setting Up SparkMeasure with Spark
  • Version Compatibility for SparkMeasure
  • 📥 Downloading SparkMeasure
  • Setup Examples
    • Spark 4 with Scala 2.13
    • Spark 3 with Scala 2.12
  • Including sparkMeasure in your Spark environment
  • Running unit tests
  • Notes on Spark Metrics
• Documentation, API, and examples
  • Architecture diagram
  • Main concepts underlying sparkMeasure implementation

Links to related work on Spark Performance

• Building an Apache Spark Performance Lab Guide to setting up a Spark performance testing environment.
• TPC-DS Benchmark with PySpark Tool for running TPC-DS with PySpark, instrumented with sparkMeasure.
• Spark Monitoring Dashboard Custom monitoring solution with real-time dashboards for Spark.
• Flamegraphs for Profiling Spark Jobs Guide to profiling Spark with Pyroscope and Flamegraphs.
• Advanced Notes on Apache Spark Tips, configuration, and troubleshooting for Spark.
• Introductory Course on Apache Spark Beginner-friendly course on Spark fundamentals.

Main author and contact: Luca.Canali@cern.ch

---

🚀 Quick start

Watch sparkMeasure's getting started demo tutorial

Examples of sparkMeasure on notebooks

• Run locally or on hosted resources like Google Colab, Databricks, GitHub Codespaces, etc on Jupyter notebooks

• Jupyter notebook on Google Colab Research

• Scala notebook on Databricks

• Python notebook on Databricks

• Local Python/Jupyter Notebook

Examples of sparkMeasure on the CLI

• Run locally or on hosted resources
  • 

Python CLI

# Python CLI
# pip install pyspark
pip install sparkmeasure
pyspark --packages ch.cern.sparkmeasure:spark-measure_2.13:0.28

# Import sparkMeasure
from sparkmeasure import StageMetrics
stagemetrics = StageMetrics(spark)
# Simple one-liner to run a Spark SQL query and measure its performance
stagemetrics.runandmeasure(globals(), 'spark.sql("select count(*) from range(1000) cross join range(1000) cross join range(1000)").show()')

# Alternatively, you can use the begin() and end() methods to measure performance
# Start measuring
stagemetrics.begin()

spark.sql("select count(*) from range(1000) cross join range(1000) cross join range(1000)").show()

# Set a stop point for measuring metrics delta values
stagemetrics.end()
# Print the metrics report
stagemetrics.print_report()
stagemetrics.print_memory_report()

# get metrics as a dictionary
metrics = stagemetrics.aggregate_stage_metrics()

Note: for Spark 3.x with Scala 2.12, use --packages ch.cern.sparkmeasure:spark-measure_2.12:0.28 instead of --packages ch.cern.sparkmeasure:spark-measure_2.13:0.28

Scala CLI

spark-shell --packages ch.cern.sparkmeasure:spark-measure_2.13:0.28

val stageMetrics = ch.cern.sparkmeasure.StageMetrics(spark)
stageMetrics.runAndMeasure(spark.sql("select count(*) from range(1000) cross join range(1000) cross join range(1000)").show())

The output should look like this:

+----------+
|  count(1)|
+----------+
|1000000000|
+----------+

Time taken: 3833 ms

Scheduling mode = FIFO
Spark Context default degree of parallelism = 8

Aggregated Spark stage metrics:
numStages => 3
numTasks => 17
elapsedTime => 1112 (1 s)
stageDuration => 864 (0.9 s)
executorRunTime => 3358 (3 s)
executorCpuTime => 2168 (2 s)
executorDeserializeTime => 892 (0.9 s)
executorDeserializeCpuTime => 251 (0.3 s)
resultSerializationTime => 72 (72 ms)
jvmGCTime => 0 (0 ms)
shuffleFetchWaitTime => 0 (0 ms)
shuffleWriteTime => 36 (36 ms)
resultSize => 16295 (15.9 KB)
diskBytesSpilled => 0 (0 Bytes)
memoryBytesSpilled => 0 (0 Bytes)
peakExecutionMemory => 0
recordsRead => 2000
bytesRead => 0 (0 Bytes)
recordsWritten => 0
bytesWritten => 0 (0 Bytes)
shuffleRecordsRead => 8
shuffleTotalBlocksFetched => 8
shuffleLocalBlocksFetched => 8
shuffleRemoteBlocksFetched => 0
shuffleTotalBytesRead => 472 (472 Bytes)
shuffleLocalBytesRead => 472 (472 Bytes)
shuffleRemoteBytesRead => 0 (0 Bytes)
shuffleRemoteBytesReadToDisk => 0 (0 Bytes)
shuffleBytesWritten => 472 (472 Bytes)
shuffleRecordsWritten => 8

Average number of active tasks => 3.0

Stages and their duration:
Stage 0 duration => 355 (0.4 s)
Stage 1 duration => 411 (0.4 s)
Stage 3 duration => 98 (98 ms)

Memory report

Stage metrics collection mode has an optional memory report command:

(scala)> stageMetrics.printMemoryReport
(python)> stagemetrics.print_memory_report()

Additional stage-level executor metrics (memory usage info updated at each heartbeat):

Stage 0 JVMHeapMemory maxVal bytes => 322888344 (307.9 MB)
Stage 0 OnHeapExecutionMemory maxVal bytes => 0 (0 Bytes)
Stage 1 JVMHeapMemory maxVal bytes => 322888344 (307.9 MB)
Stage 1 OnHeapExecutionMemory maxVal bytes => 0 (0 Bytes)
Stage 3 JVMHeapMemory maxVal bytes => 322888344 (307.9 MB)
Stage 3 OnHeapExecutionMemory maxVal bytes => 0 (0 Bytes)

Notes:

• this report makes use of per-stage memory (executor metrics) data which is sent by the executors at each heartbeat to the driver, there could be a small delay or the order of a few seconds between the end of the job and the time the last metrics value is received.

CLI example for Task Metrics:

This is similar but slightly different from the example above as it collects metrics at the Task-level rather than Stage-level

# Scala CLI
spark-shell --packages ch.cern.sparkmeasure:spark-measure_2.13:0.28

val taskMetrics = ch.cern.sparkmeasure.TaskMetrics(spark)
taskMetrics.runAndMeasure(spark.sql("select count(*) from range(1000) cross join range(1000) cross join range(1000)").show())

# Python CLI
# pip install pyspark
pip install sparkmeasure
pyspark --packages ch.cern.sparkmeasure:spark-measure_2.13:0.28

from sparkmeasure import TaskMetrics
taskmetrics = TaskMetrics(spark)
taskmetrics.runandmeasure(globals(), 'spark.sql("select count(*) from range(1000) cross join range(1000) cross join range(1000)").show()')

---

Setting Up SparkMeasure with Spark

Version Compatibility for SparkMeasure

Spark Version	Recommended SparkMeasure Version	Scala Version
Spark 4.x	0.28 (latest)	Scala 2.13
Spark 3.x	0.28 (latest)	Scala 2.12 and 2.13
Spark 2.4, 2.3	0.19	Scala 2.11
Spark 2.2, 2.1	0.16	Scala 2.11

📥 Downloading SparkMeasure

To get SparkMeasure, choose one of the following options:

1. Stable Releases:

   • Available on Maven Central Repository.

2. Specific Versions:

   • Download JAR files from the sparkMeasure release notes.

3. Latest Development Builds:

   • Access the latest development builds from [GitHub Actions](https://github.com/LucaCanali/sparkMeasur