PySpark Overview - Part 3.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Overview\n",
    "\n",
    "In this notebook, I want to cover areas related to performance as well as some more advanced spark functionality (MLib, Streaming, DataFrames). This notebook is to give you a taste of what is possible with spark, and is **not** a comprehensive guide.\n",
    "\n",
    "## Terminology\n",
    "\n",
    "Let's get some key terminology down first before we delve into the nuts and bolts of performance enhancements. I will motivate this with a simple example:\n",
    "\n",
    "### Spark Job\n",
    "\n",
    "Whenever we call an action (i.e. `collect()`), this action always returns a result. As such, we can say that this action resulted in a **spark job**.\n",
    "\n",
    "Each **spark job** is comprised of a set of **stages**. It is important to note here that the **number of stages in a job is directly related to the number of shuffle operations required to deliver a result.**\n",
    "\n",
    "### Stages\n",
    "\n",
    "A stage is a group of **tasks** that need to be executed in unison to compute **an operation on several machines**.\n",
    "\n",
    "Spark is optimized to put as many possible **tasks**  into a single stage, however, a **new stage is always created** after a **shuffle** occurs.\n",
    "\n",
    "Stages are **always executed in parallel.**\n",
    "\n",
    "### Shuffle\n",
    "\n",
    "A shuffle represents a **repartitioning** of the data i.e. sorting or grouping data.\n",
    "\n",
    "When we run these types of operations, the executors must co-ordinate in order for the data to move around in the appropriate order.\n",
    "\n",
    "Spark is also optimized to keep track of the stages and the order in which they must be executed in order to complete the spark job.\n",
    "\n",
    "### Tasks\n",
    "\n",
    "As stated before, a stage consists of various **tasks**. A **task** is simply a combination of chunks of data and a series of transformations that will be run on a **single executor.**\n",
    "\n",
    "You can think of a task as a **unit of computation** that is applied to a **unit of data**. In this context, the unit of data is a **partition**.\n",
    "\n",
    "As such, if we have 8 partitions, we would have have 8 tasks.\n",
    "\n",
    "Furthermore, we previously states that **stages are executed in parallel.** As such, the higher we set our number of partitions, the higher parallelism we will achieve.\n",
    "\n",
    "Recall, the spark documentation recommends 2-4 CPUs per core in your cluster."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Performance\n",
    "\n",
    "### Broadcast Variables\n",
    "\n",
    "Let's imagine for a moment that we are dealing with a very large dataset. \n",
    "\n",
    "Furthermore, let's say that we need all of our executors to access this dataset.\n",
    "\n",
    "Now, we could do this by simply assigning the data to variable, and then pass this variable around:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "really_large_dataset = 100"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize(range(10))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[100, 101, 104, 109, 116, 125, 136, 149, 164, 181]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.map(lambda x: x**2 + really_large_dataset).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, the above code is perfectly correct. Simply referencing your large data set when you have a singular stage to process is no problem, but what if we had multiple stages:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[10100, 10301, 10916, 11981, 13556, 15725, 18596, 22301, 26996, 32861]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.map(lambda x: x**2 + really_large_dataset).repartition(4).map(\n",
    "lambda x: x ** 2 + really_large_dataset).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Again, the above code might look fine to you. However, the above code is **masking** a very important consideration.\n",
    "\n",
    "The variable `really_large_dataset`, is being **serialized** and sent with the task information for **both** stages.\n",
    "\n",
    "Now, the above example is incredibly simplistic, but if we really did have a large amount of data, **numerous serializations** is highly inefficient.\n",
    "\n",
    "Now, since sharing data between executors is so common, spark has a way to **efficiently** handle this. Instead of just referencing a variable, we use **broadcast variables.**\n",
    "\n",
    "In essence, a broadcast variable ensures that your data remains available to all executors **between stages**. The data is simple sent to the spark job once, instead of as many times as there are stages.\n",
    "\n",
    "Here is an example of a broadcast variable:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# this data is now sent out to the executors.\n",
    "really_large_dataset = sc.broadcast(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.broadcast.Broadcast at 0x1094bf080>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "really_large_dataset #broadcast variable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "100"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "really_large_dataset.value #Broadcast variable VALUE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[100, 101, 104, 109, 116, 125, 136, 149, 164, 181]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#same result as before!\n",
    "rdd.map(lambda x: x**2 + really_large_dataset.value).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Broadcast Variable vs Broadcast Value"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we see can see from above, we obtain the same results as before when using a broadcast variable. However, we used the `.value` attribute.\n",
    "\n",
    "This defeats the purpose of using a broadcast variable. Rather, we should just pass around the broadcast variable and extract the value at the very end. This will ensure that we are making good use of our efficiency gains.\n",
    "\n",
    "#### Read Only\n",
    "\n",
    "Broadcast variables are read only. They cannot be changed in a particular place and expect all executors to be made aware of the change.\n",
    "\n",
    "It is good practice to set your broadcast variables up at the very beginning and leave them unchanged. Attempts to change their value in the executors will lead to very odd results.\n",
    "\n",
    "#### Persistence\n",
    "\n",
    "One of the side effects of broadcast variables is that spark creates a cached copy of the data. This is what allows the data to be available to all executors between stages.\n",
    "\n",
    "As such, we you are done with a particular broadcast variable, be sure to remove cached copies to ensure continued efficiency of your code. Unpersisting the data will allow you to reclaim memory space."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "really_large_dataset.unpersist()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Accumulators\n",
    "\n",
    "For those of you experienced with functional programming, accumulators will be nothing new. Haskell constantly uses accumulators.\n",
    "\n",
    "In essence, an accumulator is a structure that can **always be added to** but is **only visible** in the **main driver.**\n",
    "\n",
    "Let's illustrate this with a simple example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "even_counter = sc.accumulator(0)\n",
    "\n",
    "data = sc.parallelize(range(20))\n",
    "\n",
    "def add_one(val):\n",
    "    global even_counter\n",
    "    if (val%2 == 0):\n",
    "        even_counter += 1\n",
    "    return val + 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# results \n",
    "data.map(add_one).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# value of the accumulator - the number of even numbers in data\n",
    "even_counter.value"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, the above code is pretty self explanatory, however, lets talk about some common mistakes that occur with accumulators.\n",
    "\n",
    "#### Running actions numerous times\n",
    "\n",
    "If I were to run `data.map(add_one).collect()` again, our `even_counter` would now have a value of 20.\n",
    "\n",
    "Recall that we did **not** reset the accumulator value to 0. As such, for every action, our accumulator value will update.\n",
    "\n",
    "#### Accumulators & transformations/failures\n",
    "\n",
    "There are times where a spark job will fail - in these instances, spark will sometimes restart the job. When this happens, it is possible that action is executed multiple times. This will result in your accumulator value incrementing a number of times.\n",
    "\n",
    "Furthermore, when spark is running transformations, there is no guarantee that accumulators will be consistent.\n",
    "\n",
    "\n",
    "Thus, counting inside of an **action** has the guarantee of hitting your accumulator once. Counting inside of a **transformation** has no guarantee of hitting your accumulator once.\n",
    "\n",
    "#### Lazy Evaluation\n",
    "\n",
    "Recall that spark relies on lazy evaluation, thus if I had just run `data.map(add_one)`, our counter would have remained at 0.\n",
    "\n",
    "Accumulators are often used for internal statistics or maintaining long pipelines.\n",
    "\n",
    "They are a very useful feature!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Advanced Spark\n",
    "\n",
    "### Spark Streaming\n",
    "\n",
    "Spark Streaming is the abstraction required for dealing with data in real-time. It allows your to read in input data in discrete time chunks and run various processes on those chunks.\n",
    "\n",
    "This is incredibly important for scalable/production grade solutions using spark architecture.\n",
    "\n",
    "There is incredibly detailed and well written documentation available.\n",
    "\n",
    "Below I will give a very simple example.\n",
    "\n",
    "The central construct in spark streaming is not an RDD, but rather [Discretized Stream (DSteams)](https://spark.apache.org/docs/2.2.0/streaming-programming-guide.html#discretized-streams-dstreams)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.streaming import StreamingContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# the 5 means we will stream every 5 seconds!\n",
    "stream_context = StreamingContext(sc, 5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "input_file = stream_context.textFileStream(\"test\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pairs = input_file.flatMap(lambda x: x.split(\" \")).map(lambda v: (v,1))\n",
    "\n",
    "count = pairs.reduceByKey(lambda x,y: x+y)\n",
    "\n",
    "count.pprint()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2018-02-25 20:17:00\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-02-25 20:17:05\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-02-25 20:17:10\n",
      "-------------------------------------------\n",
      "('B', 1)\n",
      "('C', 1)\n",
      "('A', 1)\n",
      "('words', 1)\n",
      "('some', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-02-25 20:17:15\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-02-25 20:17:20\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-02-25 20:17:25\n",
      "-------------------------------------------\n",
      "\n"
     ]
    }
   ],
   "source": [
    "stream_context.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "stream_context.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "What is important to note in the above example is that I **first** started the stream and **then created a file on the fly** in the `test` directory.\n",
    "\n",
    "This mimics the arrival of new data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### DataFrame & SQL\n",
    "\n",
    "Spark's API allows you to interface with DataFrames (think Pandas, R) as well as SQL.\n",
    "\n",
    "Let's walk through an example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)\n",
    "data = sqlContext.read.json(\"samples/sample.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Height: double (nullable = true)\n",
      " |-- id: long (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#show the inferred schema SQL style\n",
    "data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---+-------+\n",
      "|Height| id|   name|\n",
      "+------+---+-------+\n",
      "| 181.5|  1|Ibrahim|\n",
      "| 175.4|  2|   Juan|\n",
      "| 160.7|  3| Andrew|\n",
      "+------+---+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# see data in tabular form!\n",
    "data.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, there are 2 ways to run SQL queries. The first way is by applying SQL methods to `data`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  1|\n",
      "|  2|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.filter(data['Height'] > 161).select(data['id']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The second way is to write SQL queries directly! For this, we need to use the `registerTempTable` method first!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# I am giving our data a name and instantiating a SQL table\n",
    "data.registerTempTable(\"my_data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(id=1), Row(id=2)]"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT id FROM my_data WHERE Height > 161\").collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark [DataFrames](https://spark.apache.org/docs/2.2.0/sql-programming-guide.html#datasets-and-dataframes) are used extensively for Machine Learning. Be sure to check them out in the documentation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ML & MLib\n",
    "\n",
    "There are 2 machine learning API's for spark, ML & MLib:\n",
    "\n",
    "- [ML](https://spark.apache.org/docs/2.2.0/ml-guide.html): A higher level API that is built upon **DataFrames**\n",
    "\n",
    "- [MLib](https://spark.apache.org/docs/2.2.0/mllib-guide.html): A lower level API that is built upon **RDDs**\n",
    "\n",
    "Where possible, you should always aim to use **ML** - DataFrames are buit upon RDDs. However, DataFrames should always be the data structure of choice when using spark, especially in the context of machine learning, unless you really need to fidelity of an RDD.\n",
    "\n",
    "The ML API uses the abstraction of a Pipeline - click [here](https://spark.apache.org/docs/2.2.0/ml-pipeline.html#example-pipeline) to see an excellent example of this in the documentation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Conclusion\n",
    "\n",
    "That's all I have for this short series on on PySpark!\n",
    "\n",
    "In essence, this series focused on the [RDD](https://spark.apache.org/docs/2.2.0/rdd-programming-guide.html#resilient-distributed-datasets-rdds) which is at the core of all spark abstractions!\n",
    "\n",
    "In practice, it is rare that you will always be manipulating RDD's - but understanding how RDDs work are essential when thinking about how to write efficient code.\n",
    "\n",
    "It is more likely that you will interface with [DataFrames](https://spark.apache.org/docs/2.2.0/sql-programming-guide.html#datasets-and-dataframes) in your day to day spark'ing. The documentation is pretty excellent for this data structure. You should have everything you need to shift over to dataframes! Especially if you have any experience with Pandas/R.\n",
    "\n",
    "I would highly recommend the following [book](https://www.amazon.com/Spark-Definitive-Guide-Processing-Simple/dp/1491912219) for those who want to continue their spark journey.\n",
    "\n",
    "This blog post was far from a comprehensive overview - it is important to keep in mind that Spark is an **active** project and this will change over time. Rather, I hope this blog series gave you an insight into thinking about code in the context of big data as well as a flavor of the tools, abstractions and methods that can be used to streamline your work flow in addition to generating insight along the way.\n",
    "\n",
    "As always, feel free to get in touch: igabr@uchicago.edu or [@Gabr\\_Ibrahim](https://twitter.com/Gabr_Ibrahim)\n",
    "\n",
    "Cheers!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}