This repository contains the development code for sparkMeasure, an Apache Spark performance analysis and troubleshooting library. It simplifies collecting, aggregating, and exporting Spark task/stage metrics, and is designed for practical use by developers and data engineers in interactive analysis, testing, and production monitoring workflows.
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
- Development & CI/CD Integration: Facilitates testing, measuring, and comparing execution metrics
- Batch Job Analysis: With Flight Recorder mode sparkMeasure transparently records batch job metrics
- Monitoring Capabilities: Integrates with external systems like Apache Kafka,
- Educational Tool: Serves as a practical example of implementing Spark Listeners for the collection
- Language Compatibility: Fully supports Scala, Java, and Python, making it versatile for a wide range
📚 Table of Contents
- ✨ Highlights - 📚 Table of Contents - Links to related work on Spark Performance - Examples of sparkMeasure on notebooks - Examples of sparkMeasure on the CLI - Python CLI - Scala CLI - Memory report - CLI example for Task Metrics: - 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 - Architecture diagram - Main concepts underlying sparkMeasure implementation - FAQ:Links to related work on Spark Performance
Guide to setting up a Spark performance testing environment. Tool for running TPC-DS with PySpark, instrumented withsparkMeasure.
Custom monitoring solution with real-time dashboards for Spark.
Guide to profiling Spark with Pyroscope and Flamegraphs.
Tips, configuration, and troubleshooting for 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
Local Python/Jupyter NotebookExamples 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.printmemoryreport()
# get metrics as a dictionary metrics = stagemetrics.aggregatestagemetrics()
--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.printmemoryreport()
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)
- this report makes use of per-stage memory (executor metrics) data which is sent by the
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:
- Stable Releases:
- Specific Versions:
- Latest Development Builds:
- Build from Source:
sbt +package.
Setup Examples
Spark 4 with Scala 2.13
- Scala:
spark-shell --packages ch.cern.sparkmeasure:spark-measure_2.13:0.28 - Python:
pyspark --packages ch.cern.sparkmeasure:spark-measure_2.13:0.28
pip install sparkmeasureSpark 3 with Scala 2.12
- Scala:
spark-shell --packages ch.cern.sparkmeasure:spark-measure_2.12:0.28 - Python:
pyspark --packages ch.cern.sparkmeasure:spark-measure_2.12:0.28
pip install sparkmeasureAdd sparkMeasure to your Spark runtime classpath
Choose one of the following options. You only need a single method to make sparkMeasure available to Spark.
- Preferred: use the
--packagesoption to download sparkMeasure from Maven Central and automatically resolve its dependencies:
--packages ch.cern.sparkmeasure:spark-measure_2.13:0.28- Alternatively, provide the JAR yourself with one of these direct JAR/classpath methods:
--jars /path/to/spark-measure_2.13-0.28.jar
--jars https://github.com/LucaCanali/sparkMeasure/releases/download/v0.28/spark-measure_2.13-0.28.jar
--conf spark.driver.extraClassPath=/path/to/spark-measure_2.13-0.28.jarRunning unit tests
To ensure the integrity of the sparkMeasure codebase and validate your setup, you can run the built-in unit tests. These tests are designed to verify core functionality. Running sparkMeasure unit tests
Notes on Spark Metrics
Spark is instrumented with several metrics, collected at task execution, they are described in the documentation: Some of the key metrics when looking at a sparkMeasure report are:- elapsedTime: the time taken by the stage or task to complete (in millisec)
- executorRunTime: the time the executors spent running the task, (in millisec). Note this time is cumulative across all tasks executed by the executor.
- executorCpuTime: the time the executors spent running the task, (in millisec). Note this time is cumulative across all tasks executed by the executor.
- jvmGCTime: the time the executors spent in garbage collection, (in millisec).
- shuffle metrics: several metrics with details on the I/O and time spend on shuffle
- I/O metrics: details on the I/O (reads and writes). Note, currently there are no time-based metrics for I/O operations.
Documentation, API, and examples
SparkMeasure is one tool for many different use cases, languages, and environments: *
- SparkMeasure's API and configurations
* Interactive mode Use sparkMeasure to collect and analyze Spark workload metrics in interactive mode when working with shell or notebook environments, such as spark-shell (Scala), PySpark (Python) and/or from jupyter notebooks. - SparkMeasure on PySpark and Jupyter notebooks - SparkMeasure on Scala shell and notebooks
* Batch and code instrumentation Instrument your code with the sparkMeasure API, for collecting, saving, and analyzing Spark workload metrics data. Examples and how-to guides: - Instrument Spark-Python code - Instrument Spark-Scala code - JMX Exporter integration - See also SparkCPUmemoryloadtestkit as an example of how to use sparkMeasure to instrument Spark code for performance testing.
* Flight Recorder mode SparkMeasure in flight recorder will collect metrics transparently, without any need for you to change your code. * Metrics can be saved to a file, locally, or to a Hadoop-compliant filesystem * or you can write metrics in near-realtime to the following sinks: InfluxDB, Apache Kafka, Prometheus PushPushgateway * More details: - Flight Recorder mode with file sink - Flight Recorder mode with InfluxDB sink - Flight Recorder mode with Apache Kafka sink - Flight Recorder mode with Prometheus Pushgateway sink
* Limitations and known issues * Support for Spark Connect SparkMeasure cannot yet provide full integration with Spark Connect because it needs direct access to the SparkContext and its listener interface. You can run sparkMeasure in Flight Recorder mode on the Spark Connect drive* to capture metrics for the entire application, but per-client (Spark Connect client-side) metrics are not reported. * Single-threaded applications The sparkMeasure APIs for generating reports using StageMetric and TaskMetric are best suited for a single-threaded driver environment. These APIs capture metrics by calculating deltas between snapshots taken at the start and end of an execution. If multiple Spark actions run concurrently on the Spark driver, it may result in double-counting of metric values.
* Additional documentation and examples - Presentations at Spark/Data+AI Summit: - Performance Troubleshooting Using Apache Spark Metrics - Apache Spark Performance Troubleshooting at Scale, Challenges, Tools, and Methodologies - Blog articles: - 2018: SparkMeasure, a tool for performance troubleshooting of Apache Spark workloads, - 2017: SparkMeasure blog post - TODO list and known issues - TPCDS-PySpark a tool for running the TPCDS benchmark workload with PySpark and instrumented with sparkMeasure
Architecture diagram
Main concepts underlying sparkMeasure implementation
- The tool is based on the Spark Listener interface. Listeners transport Spark executor
- The tool is built on multiple modules implemented as classes
- Metrics are flattened and collected into local memory structures in the driver (ListBuffer of a custom case class).
- Metrics processing:
- Metrics data and reports can be saved for offline analysis.
FAQ:
- Why measuring performance with workload metrics instrumentation rather than just using execution time measurements? - When measuring just the jobs' elapsed time, you treat your workload as "a black box" and most often this does not allow you to understand the root causes of performance regression. With workload metrics you can (attempt to) go further in understanding and perform root cause analysis, bottleneck identification, and resource usage measurement.- What are Apache Spark task metrics and what can I use them for? - Apache Spark measures several details of each task execution, including run time, CPU time, information on garbage collection time, shuffle metrics, and task I/O. See also Spark documentation for a description of the Spark Task Metrics
- How is sparkMeasure different from Web UI/Spark History Server and EventLog? - sparkMeasure uses the same ListenerBus infrastructure used to collect data for the Web UI and Spark EventLog. - Spark collects metrics and other execution details and exposes them via the Web UI. - Notably, Task execution metrics are also available through the REST API - In addition, Spark writes all details of the task execution in the EventLog file (see config of spark.eventlog.enabled and spark.eventLog.dir) - The EventLog is used by the Spark History server + other tools and programs that can read and parse the EventLog file(s) for workload analysis and performance troubleshooting, see a proof-of-concept example of reading the EventLog with Spark SQL - There are key differences that motivate this development: - sparkMeasure can collect data at the stage completion-level, which is more lightweight than measuring all the tasks, in case you only need to compute aggregated performance metrics. When needed, sparkMeasure can also collect data at the task granularity level. - sparkMeasure has an API that makes it simple to add instrumentation/performance measurements in notebooks and in application code for Scala, Java, and Python. - sparkMeasure collects data in a flat structure, which makes it natural to use Spark SQL for workload data analysis/ - sparkMeasure can sink metrics data into external systems (Filesystem, InfluxDB, Apache Kafka, Prometheus Pushgateway)
- What are known limitations and gotchas? - sparkMeasure does not collect all the data available in the EventLog - See also the TODO and issues doc - The currently available Spark task metrics can give you precious quantitative information on resources used by the executors, however there do not allow to fully perform time-based analysis of the workload performance, notably they do not expose the time spent doing I/O or network traffic. - Metrics are collected on the driver, which could become a bottleneck. This is an issues affecting tools based on Spark ListenerBus instrumentation, such as the Spark WebUI. In addition, note that sparkMeasure in the current version buffers all data in the driver memory. The notable exception is when using the Flight recorder mode with InfluxDB or Apache Kafka or Prometheus Pushgateway sink, in this case metrics are directly sent to InfluxDB/Kafka/Prometheus Pushgateway. - Task metrics values are collected by sparkMeasure only for successfully executed tasks. Note that resources used by failed tasks are not collected in the current version. The notable exception is with the Flight recorder mode with InfluxDB or Apache Kafka or Prometheus Pushgateway sink. - sparkMeasure collects and processes data in order of stage and/or task completion. This means that the metrics data is not available in real-time, but rather with a delay that depends on the workload and the size of the data. Moreover, performance data of jobs executing at the same time can be mixed. This can be a noticeable issue if you run workloads with many concurrent jobs. - Task metrics are collected by Spark executors running on the JVM, resources utilized outside the JVM are currently not directly accounted for (notably the resources used when running Python code inside the python.daemon in the case of Python UDFs with PySpark).
- When should I use Stage-level metrics and when should I use Task-level metrics? - Use stage metrics whenever possible as they are much more lightweight. Collect metrics at the task granularity if you need the extra information, for example if you want to study effects of skew, long tails and task stragglers.
- How can I save/sink the collected metrics? - You can print metrics data and reports to standard output or save them to files, using a locally mounted filesystem or a Hadoop compliant filesystem (including HDFS). Additionally, you can sink metrics to external systems (such as Prometheus Pushgateway). The Flight Recorder mode can sink to InfluxDB, Apache Kafka or Prometheus Pushgateway.
- How can I process metrics data? - You can use Spark to read the saved metrics data and perform further post-processing and analysis. See the also Notes on metrics analysis.
- How can I contribute to sparkMeasure? - SparkMeasure has already profited from users submitting PR contributions. Additional contributions are welcome. See the TODOand_issues list for a list of known issues and ideas on what you can contribute.
