examples/SparkMeasure_Jupyter_Python_getting_started.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Example Jupyter notebook to showcase sparkMeasure APIs for Python\n",
    "  \n",
    "**SparkMeasure is a tool for performance troubleshooting of Apache Spark workloads**  \n",
    "It simplifies the collection and analysis of Spark performance metrics. It is also intended as a working example of how to use Spark listeners for collecting and processing Spark executors task metrics data.\n",
    "\n",
    "**References:**\n",
    "- [https://github.com/LucaCanali/sparkMeasure](https://github.com/LucaCanali/sparkMeasure)  \n",
    "- sparkmeasure Python docs: [docs/Python_shell_and_Jupyter](https://github.com/LucaCanali/sparkMeasure/blob/master/docs/Python_shell_and_Jupyter.md)  \n",
    "\n",
    "**Architecture:**\n",
    "![sparkMeasure architecture diagram](https://github.com/LucaCanali/sparkMeasure/raw/master/docs/sparkMeasure_architecture_diagram.png)\n",
    "\n",
    "Author and contact: Luca.Canali@cern.ch  \n",
    "Last updated: March 2024"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dependencies\n",
    "\n",
    "# 1. Install PySpark/Spark \n",
    "# This is optional, if you have already downloaded Spark\n",
    "# See findspark (3.) in that case\n",
    "!pip install pyspark\n",
    "\n",
    "# 2. Install the Python wrapper API for spark-measure\n",
    "!pip install sparkmeasure\n",
    "\n",
    "# 3. The use of findspark is optional\n",
    "# It can be handy if you have to choose among multiple Spark homes\n",
    "# !pip install findspark\n",
    "# import findspark\n",
    "# findspark.init(\"/home/luca/Spark/spark-3.5.1-bin-hadoop3\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start the Spark Session\n",
    "# This example uses Spark in local mode for simplicity.\n",
    "# You can modify master to use  YARN or K8S if available \n",
    "# This example uses sparkMeasure 0.24 for scala 2.12, taken from maven central\n",
    "\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "spark = (SparkSession.builder\n",
    "         .appName(\"Test sparkmeasure instrumentation of Python/PySpark code\")\n",
    "         .master(\"local[*]\")\n",
    "         .config(\"spark.jars.packages\",\"ch.cern.sparkmeasure:spark-measure_2.12:0.24\")\n",
    "         .getOrCreate()\n",
    "        )  \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initialize sparkMeasure\n",
    "# Load the Python API for sparkmeasure package\n",
    "# and attach the sparkMeasure Listener for stagemetrics to the active Spark session\n",
    "\n",
    "from sparkmeasure import StageMetrics\n",
    "stagemetrics = StageMetrics(spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  count(1)|\n",
      "+----------+\n",
      "|1000000000|\n",
      "+----------+\n",
      "\n",
      "\n",
      "Scheduling mode = FIFO\n",
      "Spark Context default degree of parallelism = 8\n",
      "\n",
      "Aggregated Spark stage metrics:\n",
      "numStages => 3\n",
      "numTasks => 17\n",
      "elapsedTime => 1151 (1 s)\n",
      "stageDuration => 936 (0.9 s)\n",
      "executorRunTime => 3255 (3 s)\n",
      "executorCpuTime => 2116 (2 s)\n",
      "executorDeserializeTime => 909 (0.9 s)\n",
      "executorDeserializeCpuTime => 228 (0.2 s)\n",
      "resultSerializationTime => 36 (36 ms)\n",
      "jvmGCTime => 0 (0 ms)\n",
      "shuffleFetchWaitTime => 0 (0 ms)\n",
      "shuffleWriteTime => 11 (11 ms)\n",
      "resultSize => 16295 (15.9 KB)\n",
      "diskBytesSpilled => 0 (0 Bytes)\n",
      "memoryBytesSpilled => 0 (0 Bytes)\n",
      "peakExecutionMemory => 0\n",
      "recordsRead => 2000\n",
      "bytesRead => 0 (0 Bytes)\n",
      "recordsWritten => 0\n",
      "bytesWritten => 0 (0 Bytes)\n",
      "shuffleRecordsRead => 8\n",
      "shuffleTotalBlocksFetched => 8\n",
      "shuffleLocalBlocksFetched => 8\n",
      "shuffleRemoteBlocksFetched => 0\n",
      "shuffleTotalBytesRead => 472 (472 Bytes)\n",
      "shuffleLocalBytesRead => 472 (472 Bytes)\n",
      "shuffleRemoteBytesRead => 0 (0 Bytes)\n",
      "shuffleRemoteBytesReadToDisk => 0 (0 Bytes)\n",
      "shuffleBytesWritten => 472 (472 Bytes)\n",
      "shuffleRecordsWritten => 8\n",
      "\n",
      "Average number of active tasks => 2.8\n",
      "\n",
      "Stages and their duration:\n",
      "Stage 0 duration => 394 (0.4 s)\n",
      "Stage 1 duration => 458 (0.5 s)\n",
      "Stage 3 duration => 84 (84 ms)\n"
     ]
    }
   ],
   "source": [
    "# The easiest way to start using sparkMesure is with the \"runandmeasure\" method\n",
    "# This will execute your Spark action, return the results, and collect and aggregate execution metrics\n",
    "\n",
    "stagemetrics.runandmeasure(globals(), \"\"\"\n",
    "spark.sql(\"select count(*) from range(1000) cross join range(1000) cross join range(1000)\").show()\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Additional stage-level executor metrics (memory usage info):\n",
      "\n",
      "Stage 0 JVMHeapMemory maxVal bytes => 337096704 (321.5 MB)\n",
      "Stage 0 OnHeapExecutionMemory maxVal bytes => 0 (0 Bytes)\n",
      "Stage 1 JVMHeapMemory maxVal bytes => 337096704 (321.5 MB)\n",
      "Stage 1 OnHeapExecutionMemory maxVal bytes => 0 (0 Bytes)\n",
      "Stage 3 JVMHeapMemory maxVal bytes => 337096704 (321.5 MB)\n",
      "Stage 3 OnHeapExecutionMemory maxVal bytes => 0 (0 Bytes)\n"
     ]
    }
   ],
   "source": [
    "# Additionally, sparkMeasure collects executor metrics\n",
    "# This is how you can print the memory usage report\n",
    "# Note, If you receive the error message java.util.NoSuchElementException: key not found, \n",
    "# retry running the report after waiting for a few seconds.\n",
    "\n",
    "stagemetrics.print_memory_report()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  count(1)|\n",
      "+----------+\n",
      "|1000000000|\n",
      "+----------+\n",
      "\n",
      "\n",
      "Scheduling mode = FIFO\n",
      "Spark Context default degree of parallelism = 8\n",
      "\n",
      "Aggregated Spark stage metrics:\n",
      "numStages => 3\n",
      "numTasks => 17\n",
      "elapsedTime => 537 (0.5 s)\n",
      "stageDuration => 448 (0.4 s)\n",
      "executorRunTime => 2331 (2 s)\n",
      "executorCpuTime => 1742 (2 s)\n",
      "executorDeserializeTime => 125 (0.1 s)\n",
      "executorDeserializeCpuTime => 64 (64 ms)\n",
      "resultSerializationTime => 8 (8 ms)\n",
      "jvmGCTime => 0 (0 ms)\n",
      "shuffleFetchWaitTime => 0 (0 ms)\n",
      "shuffleWriteTime => 10 (10 ms)\n",
      "resultSize => 16080 (15.7 KB)\n",
      "diskBytesSpilled => 0 (0 Bytes)\n",
      "memoryBytesSpilled => 0 (0 Bytes)\n",
      "peakExecutionMemory => 0\n",
      "recordsRead => 2000\n",
      "bytesRead => 0 (0 Bytes)\n",
      "recordsWritten => 0\n",
      "bytesWritten => 0 (0 Bytes)\n",
      "shuffleRecordsRead => 8\n",
      "shuffleTotalBlocksFetched => 8\n",
      "shuffleLocalBlocksFetched => 8\n",
      "shuffleRemoteBlocksFetched => 0\n",
      "shuffleTotalBytesRead => 472 (472 Bytes)\n",
      "shuffleLocalBytesRead => 472 (472 Bytes)\n",
      "shuffleRemoteBytesRead => 0 (0 Bytes)\n",
      "shuffleRemoteBytesReadToDisk => 0 (0 Bytes)\n",
      "shuffleBytesWritten => 472 (472 Bytes)\n",
      "shuffleRecordsWritten => 8\n",
      "\n",
      "Average number of active tasks => 4.3\n",
      "\n",
      "Stages and their duration:\n",
      "Stage 4 duration => 39 (39 ms)\n",
      "Stage 5 duration => 389 (0.4 s)\n",
      "Stage 7 duration => 20 (20 ms)\n"
     ]
    }
   ],
   "source": [
    "# An equivalent API for collecting execution metrics is to explicitly wrap your Spark workload\n",
    "# into stagemetrics instrumentation, as in this example\n",
    "\n",
    "stagemetrics.begin()\n",
    "\n",
    "spark.sql(\"select count(*) from range(1000) cross join range(1000) cross join range(1000)\").show()\n",
    "\n",
    "stagemetrics.end()\n",
    "\n",
    "# Print a summary report\n",
    "stagemetrics.print_report()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This code is for Jupyter notebooks\n",
    "# Define cell and line magic to wrap sparkmeasure instrumentation\n",
    "# See example in the next cell\n",
    "\n",
    "from IPython.core.magic import (register_line_magic, register_cell_magic, register_line_cell_magic)\n",
    "\n",
    "@register_line_cell_magic\n",
    "def sparkmeasure(line, cell=None):\n",
    "    \"run and measure spark workload. Use: %sparkmeasure or %%sparkmeasure\"\n",
    "    val = cell if cell is not None else line\n",
    "    stagemetrics.begin()\n",
    "    eval(val)\n",
    "    stagemetrics.end()\n",
    "    stagemetrics.print_report()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  count(1)|\n",
      "+----------+\n",
      "|1000000000|\n",
      "+----------+\n",
      "\n",
      "\n",
      "Scheduling mode = FIFO\n",
      "Spark Context default degree of parallelism = 8\n",
      "\n",
      "Aggregated Spark stage metrics:\n",
      "numStages => 3\n",
      "numTasks => 17\n",
      "elapsedTime => 478 (0.5 s)\n",
      "stageDuration => 398 (0.4 s)\n",
      "executorRunTime => 2259 (2 s)\n",
      "executorCpuTime => 1752 (2 s)\n",
      "executorDeserializeTime => 78 (78 ms)\n",
      "executorDeserializeCpuTime => 48 (48 ms)\n",
      "resultSerializationTime => 31 (31 ms)\n",
      "jvmGCTime => 105 (0.1 s)\n",
      "shuffleFetchWaitTime => 0 (0 ms)\n",
      "shuffleWriteTime => 48 (48 ms)\n",
      "resultSize => 16467 (16.1 KB)\n",
      "diskBytesSpilled => 0 (0 Bytes)\n",
      "memoryBytesSpilled => 0 (0 Bytes)\n",
      "peakExecutionMemory => 0\n",
      "recordsRead => 2000\n",
      "bytesRead => 0 (0 Bytes)\n",
      "recordsWritten => 0\n",
      "bytesWritten => 0 (0 Bytes)\n",
      "shuffleRecordsRead => 8\n",
      "shuffleTotalBlocksFetched => 8\n",
      "shuffleLocalBlocksFetched => 8\n",
      "shuffleRemoteBlocksFetched => 0\n",
      "shuffleTotalBytesRead => 472 (472 Bytes)\n",
      "shuffleLocalBytesRead => 472 (472 Bytes)\n",
      "shuffleRemoteBytesRead => 0 (0 Bytes)\n",
      "shuffleRemoteBytesReadToDisk => 0 (0 Bytes)\n",
      "shuffleBytesWritten => 472 (472 Bytes)\n",
      "shuffleRecordsWritten => 8\n",
      "\n",
      "Average number of active tasks => 4.7\n",
      "\n",
      "Stages and their duration:\n",
      "Stage 8 duration => 38 (38 ms)\n",
      "Stage 9 duration => 339 (0.3 s)\n",
      "Stage 11 duration => 21 (21 ms)\n"
     ]
    }
   ],
   "source": [
    "%%sparkmeasure\n",
    "spark.sql(\"select count(*) from range(1000) cross join range(1000) cross join range(1000)\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example of using Task Metrics\n",
    "Collecting Spark task metrics at the granularity of each task completion has additional overhead\n",
    "compared to collecting at the stage-level.\n",
    "This option should only be used if you need data with this finer granularity, for example because you want\n",
    "to study skew effects, otherwise consider using stagemetrics aggregation as the preferred choice.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  count(1)|\n",
      "+----------+\n",
      "|1000000000|\n",
      "+----------+\n",
      "\n",
      "\n",
      "Scheduling mode = FIFO\n",
      "Spark Context default degree of parallelism = 8\n",
      "\n",
      "Aggregated Spark task metrics:\n",
      "numTasks => 17\n",
      "successful tasks => 17\n",
      "speculative tasks => 0\n",
      "taskDuration => 2268 (2 s)\n",
      "schedulerDelayTime => 112 (0.1 s)\n",
      "executorRunTime => 2084 (2 s)\n",
      "executorCpuTime => 1721 (2 s)\n",
      "executorDeserializeTime => 70 (70 ms)\n",
      "executorDeserializeCpuTime => 34 (34 ms)\n",
      "resultSerializationTime => 2 (2 ms)\n",
      "jvmGCTime => 0 (0 ms)\n",
      "shuffleFetchWaitTime => 0 (0 ms)\n",
      "shuffleWriteTime => 0 (0 ms)\n",
      "gettingResultTime => 0 (0 ms)\n",
      "resultSize => 4006 (3.9 KB)\n",
      "diskBytesSpilled => 0 (0 Bytes)\n",
      "memoryBytesSpilled => 0 (0 Bytes)\n",
      "peakExecutionMemory => 0\n",
      "recordsRead => 2000\n",
      "bytesRead => 0 (0 Bytes)\n",
      "recordsWritten => 0\n",
      "bytesWritten => 0 (0 Bytes)\n",
      "shuffleRecordsRead => 8\n",
      "shuffleTotalBlocksFetched => 8\n",
      "shuffleLocalBlocksFetched => 8\n",
      "shuffleRemoteBlocksFetched => 0\n",
      "shuffleTotalBytesRead => 472 (472 Bytes)\n",
      "shuffleLocalBytesRead => 472 (472 Bytes)\n",
      "shuffleRemoteBytesRead => 0 (0 Bytes)\n",
      "shuffleRemoteBytesReadToDisk => 0 (0 Bytes)\n",
      "shuffleBytesWritten => 472 (472 Bytes)\n",
      "shuffleRecordsWritten => 8\n"
     ]
    }
   ],
   "source": [
    "from sparkmeasure import TaskMetrics\n",
    "taskmetrics = TaskMetrics(spark)\n",
    "\n",
    "taskmetrics.begin()\n",
    "spark.sql(\"select count(*) from range(1000) cross join range(1000) cross join range(1000)\").show()\n",
    "taskmetrics.end()\n",
    "\n",
    "taskmetrics.print_report()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  count(1)|\n",
      "+----------+\n",
      "|1000000000|\n",
      "+----------+\n",
      "\n",
      "\n",
      "Scheduling mode = FIFO\n",
      "Spark Context default degree of parallelism = 8\n",
      "\n",
      "Aggregated Spark task metrics:\n",
      "numTasks => 17\n",
      "successful tasks => 17\n",
      "speculative tasks => 0\n",
      "taskDuration => 2320 (2 s)\n",
      "schedulerDelayTime => 108 (0.1 s)\n",
      "executorRunTime => 2128 (2 s)\n",
      "executorCpuTime => 1738 (2 s)\n",
      "executorDeserializeTime => 76 (76 ms)\n",
      "executorDeserializeCpuTime => 38 (38 ms)\n",
      "resultSerializationTime => 8 (8 ms)\n",
      "jvmGCTime => 0 (0 ms)\n",
      "shuffleFetchWaitTime => 0 (0 ms)\n",
      "shuffleWriteTime => 1 (1 ms)\n",
      "gettingResultTime => 0 (0 ms)\n",
      "resultSize => 4006 (3.9 KB)\n",
      "diskBytesSpilled => 0 (0 Bytes)\n",
      "memoryBytesSpilled => 0 (0 Bytes)\n",
      "peakExecutionMemory => 0\n",
      "recordsRead => 2000\n",
      "bytesRead => 0 (0 Bytes)\n",
      "recordsWritten => 0\n",
      "bytesWritten => 0 (0 Bytes)\n",
      "shuffleRecordsRead => 8\n",
      "shuffleTotalBlocksFetched => 8\n",
      "shuffleLocalBlocksFetched => 8\n",
      "shuffleRemoteBlocksFetched => 0\n",
      "shuffleTotalBytesRead => 472 (472 Bytes)\n",
      "shuffleLocalBytesRead => 472 (472 Bytes)\n",
      "shuffleRemoteBytesRead => 0 (0 Bytes)\n",
      "shuffleRemoteBytesReadToDisk => 0 (0 Bytes)\n",
      "shuffleBytesWritten => 472 (472 Bytes)\n",
      "shuffleRecordsWritten => 8\n"
     ]
    }
   ],
   "source": [
    "taskmetrics.runandmeasure(globals(), \"\"\"\n",
    "spark.sql(\"select count(*) from range(1000) cross join range(1000) cross join range(1000)\").show()\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}