examples/SparkMeasure_Jupyter_Python_getting_started.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Example notebook to showcase sparkMeasure APIs for Python#\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",
    "Luca.Canali@cern.ch, July 2018"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dependencies:\n",
    "    - This notebook assumes you have an active Spark sessions called \"spark\"\n",
    "    - That you have sparkmeasure jar in the driver classpath \n",
    "    - That you have installed the Python wrapper API package sparkmeasure\n",
    "\n",
    "This is some example code to help you on that:\n",
    "\n",
    "```\n",
    "# Note install sparkmeasure.py Python wrapper package if not already done:\n",
    "pip install sparkmeasure\n",
    "\n",
    "export PYSPARK_DRIVER_PYTHON=jupyter-notebook\n",
    "export PYSPARK_DRIVER_PYTHON_OPTS=\"--ip=`hostname` --no-browser\"\n",
    "# run PySpark\n",
    "bin/pyspark --packages ch.cern.sparkmeasure:spark-measure_2.11:0.14\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 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": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define cell and line magic to wrap the instrumentation\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": 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",
      "Aggregated Spark stage metrics:\n",
      "numStages => 3\n",
      "sum(numTasks) => 17\n",
      "elapsedTime => 12053 (12 s)\n",
      "sum(stageDuration) => 11950 (12 s)\n",
      "sum(executorRunTime) => 92685 (1.5 min)\n",
      "sum(executorCpuTime) => 91134 (1.5 min)\n",
      "sum(executorDeserializeTime) => 225 (0.2 s)\n",
      "sum(executorDeserializeCpuTime) => 108 (0.1 s)\n",
      "sum(resultSerializationTime) => 11 (11 ms)\n",
      "sum(jvmGCTime) => 96 (96 ms)\n",
      "sum(shuffleFetchWaitTime) => 0 (0 ms)\n",
      "sum(shuffleWriteTime) => 6 (6 ms)\n",
      "max(resultSize) => 18315 (17.0 KB)\n",
      "sum(numUpdatedBlockStatuses) => 0\n",
      "sum(diskBytesSpilled) => 0 (0 Bytes)\n",
      "sum(memoryBytesSpilled) => 0 (0 Bytes)\n",
      "max(peakExecutionMemory) => 0\n",
      "sum(recordsRead) => 2000\n",
      "sum(bytesRead) => 0 (0 Bytes)\n",
      "sum(recordsWritten) => 0\n",
      "sum(bytesWritten) => 0 (0 Bytes)\n",
      "sum(shuffleTotalBytesRead) => 472 (472 Bytes)\n",
      "sum(shuffleTotalBlocksFetched) => 8\n",
      "sum(shuffleLocalBlocksFetched) => 8\n",
      "sum(shuffleRemoteBlocksFetched) => 0\n",
      "sum(shuffleBytesWritten) => 472 (472 Bytes)\n",
      "sum(shuffleRecordsWritten) => 8\n"
     ]
    }
   ],
   "source": [
    "%%sparkmeasure\n",
    "spark.sql(\"select count(*) from range(1000) cross join range(1000) cross join range(1000)\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Aggregated Spark accumulables of type internal.metric. Sum of values grouped by metric name\n",
      "Name => sum(value) [group by name]\n",
      "\n",
      "executorCpuTime => 91135 (1.5 min)\n",
      "executorDeserializeCpuTime => 109 (0.1 s)\n",
      "executorDeserializeTime => 225 (0.2 s)\n",
      "executorRunTime => 92685 (1.5 min)\n",
      "input.recordsRead => 2000\n",
      "jvmGCTime => 96 (96 ms)\n",
      "resultSerializationTime => 11 (11 ms)\n",
      "resultSize => 30749 (30.0 KB)\n",
      "shuffle.read.fetchWaitTime => 0 (0 ms)\n",
      "shuffle.read.localBlocksFetched => 8\n",
      "shuffle.read.localBytesRead => 472 (472 Bytes)\n",
      "shuffle.read.recordsRead => 8\n",
      "shuffle.read.remoteBlocksFetched => 0\n",
      "shuffle.read.remoteBytesRead => 0 (0 Bytes)\n",
      "shuffle.read.remoteBytesReadToDisk => 0 (0 Bytes)\n",
      "shuffle.write.bytesWritten => 472 (472 Bytes)\n",
      "shuffle.write.recordsWritten => 8\n",
      "shuffle.write.writeTime => 6 (6 ms)\n",
      "\n",
      "SQL Metrics and other non-internal metrics. Values grouped per accumulatorId and metric name.\n",
      "Accid, Name => max(value) [group by accId, name]\n",
      "\n",
      "    3, data size total => 119 (119 Bytes)\n",
      "    4, duration total => 6 (6 ms)\n",
      "    5, number of output rows => 1\n",
      "    8, aggregate time total => 6 (6 ms)\n",
      "   10, duration total => 92078 (1.5 min)\n",
      "   11, number of output rows => 8\n",
      "   14, aggregate time total => 92018 (1.5 min)\n",
      "   16, number of output rows => 1000000000\n",
      "   17, number of output rows => 1000000\n",
      "   18, duration total => 92198 (1.5 min)\n",
      "   19, number of output rows => 1000\n",
      "   24, duration total => 8 (8 ms)\n",
      "   25, number of output rows => 1000\n"
     ]
    }
   ],
   "source": [
    "# Print additional metrics from accumulables\n",
    "stagemetrics.print_accumulables()"
   ]
  },
  {
   "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",
      "Aggregated Spark stage metrics:\n",
      "numStages => 3\n",
      "sum(numTasks) => 17\n",
      "elapsedTime => 13962 (14 s)\n",
      "sum(stageDuration) => 13928 (14 s)\n",
      "sum(executorRunTime) => 110345 (1.8 min)\n",
      "sum(executorCpuTime) => 109816 (1.8 min)\n",
      "sum(executorDeserializeTime) => 26 (26 ms)\n",
      "sum(executorDeserializeCpuTime) => 19 (19 ms)\n",
      "sum(resultSerializationTime) => 0 (0 ms)\n",
      "sum(jvmGCTime) => 24 (24 ms)\n",
      "sum(shuffleFetchWaitTime) => 0 (0 ms)\n",
      "sum(shuffleWriteTime) => 1 (1 ms)\n",
      "max(resultSize) => 18272 (17.0 KB)\n",
      "sum(numUpdatedBlockStatuses) => 0\n",
      "sum(diskBytesSpilled) => 0 (0 Bytes)\n",
      "sum(memoryBytesSpilled) => 0 (0 Bytes)\n",
      "max(peakExecutionMemory) => 0\n",
      "sum(recordsRead) => 2000\n",
      "sum(bytesRead) => 0 (0 Bytes)\n",
      "sum(recordsWritten) => 0\n",
      "sum(bytesWritten) => 0 (0 Bytes)\n",
      "sum(shuffleTotalBytesRead) => 472 (472 Bytes)\n",
      "sum(shuffleTotalBlocksFetched) => 8\n",
      "sum(shuffleLocalBlocksFetched) => 8\n",
      "sum(shuffleRemoteBlocksFetched) => 0\n",
      "sum(shuffleBytesWritten) => 472 (472 Bytes)\n",
      "sum(shuffleRecordsWritten) => 8\n"
     ]
    }
   ],
   "source": [
    "# You can also explicitly Wrap your Spark workload into stagemetrics instrumentation \n",
    "# as in this example\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",
    "# Print a summary report\n",
    "stagemetrics.print_report()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "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",
      "Aggregated Spark stage metrics:\n",
      "numStages => 3\n",
      "sum(numTasks) => 17\n",
      "elapsedTime => 13432 (13 s)\n",
      "sum(stageDuration) => 13394 (13 s)\n",
      "sum(executorRunTime) => 106155 (1.8 min)\n",
      "sum(executorCpuTime) => 105053 (1.8 min)\n",
      "sum(executorDeserializeTime) => 75 (75 ms)\n",
      "sum(executorDeserializeCpuTime) => 23 (23 ms)\n",
      "sum(resultSerializationTime) => 0 (0 ms)\n",
      "sum(jvmGCTime) => 0 (0 ms)\n",
      "sum(shuffleFetchWaitTime) => 0 (0 ms)\n",
      "sum(shuffleWriteTime) => 1 (1 ms)\n",
      "max(resultSize) => 17928 (17.0 KB)\n",
      "sum(numUpdatedBlockStatuses) => 0\n",
      "sum(diskBytesSpilled) => 0 (0 Bytes)\n",
      "sum(memoryBytesSpilled) => 0 (0 Bytes)\n",
      "max(peakExecutionMemory) => 0\n",
      "sum(recordsRead) => 2000\n",
      "sum(bytesRead) => 0 (0 Bytes)\n",
      "sum(recordsWritten) => 0\n",
      "sum(bytesWritten) => 0 (0 Bytes)\n",
      "sum(shuffleTotalBytesRead) => 472 (472 Bytes)\n",
      "sum(shuffleTotalBlocksFetched) => 8\n",
      "sum(shuffleLocalBlocksFetched) => 8\n",
      "sum(shuffleRemoteBlocksFetched) => 0\n",
      "sum(shuffleBytesWritten) => 472 (472 Bytes)\n",
      "sum(shuffleRecordsWritten) => 8\n"
     ]
    }
   ],
   "source": [
    "# Another way to encapsulate code and instrumentation in a compact form\n",
    "\n",
    "stagemetrics.runandmeasure(locals(), \"\"\"\n",
    "spark.sql(\"select count(*) from range(1000) cross join range(1000) cross join range(1000)\").show()\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example of collecting using Task Metrics\n",
    "Collecting Spark task metrics at the granularity of each task completion has additional overhead\n",
    "compare to collecting at the stage completion level, therefore 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 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 Contex default degree of parallelism = 8\n",
      "Aggregated Spark task metrics:\n",
      "numtasks => 17\n",
      "elapsedTime => 13406 (13 s)\n",
      "sum(duration) => 105262 (1.8 min)\n",
      "sum(schedulerDelay) => 67\n",
      "sum(executorRunTime) => 105164 (1.8 min)\n",
      "sum(executorCpuTime) => 104817 (1.7 min)\n",
      "sum(executorDeserializeTime) => 31 (31 ms)\n",
      "sum(executorDeserializeCpuTime) => 10 (10 ms)\n",
      "sum(resultSerializationTime) => 0 (0 ms)\n",
      "sum(jvmGCTime) => 0 (0 ms)\n",
      "sum(shuffleFetchWaitTime) => 0 (0 ms)\n",
      "sum(shuffleWriteTime) => 0 (0 ms)\n",
      "sum(gettingResultTime) => 0 (0 ms)\n",
      "max(resultSize) => 2241 (2.0 KB)\n",
      "sum(numUpdatedBlockStatuses) => 0\n",
      "sum(diskBytesSpilled) => 0 (0 Bytes)\n",
      "sum(memoryBytesSpilled) => 0 (0 Bytes)\n",
      "max(peakExecutionMemory) => 0\n",
      "sum(recordsRead) => 2000\n",
      "sum(bytesRead) => 0 (0 Bytes)\n",
      "sum(recordsWritten) => 0\n",
      "sum(bytesWritten) => 0 (0 Bytes)\n",
      "sum(shuffleTotalBytesRead) => 472 (472 Bytes)\n",
      "sum(shuffleTotalBlocksFetched) => 8\n",
      "sum(shuffleLocalBlocksFetched) => 8\n",
      "sum(shuffleRemoteBlocksFetched) => 0\n",
      "sum(shuffleBytesWritten) => 472 (472 Bytes)\n",
      "sum(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",
    "taskmetrics.print_report()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.15"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}