HOL04-Apache-Spark/1. Introduction to Apache Spark (MMA).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to Apache Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now one might think that this would actually print out the values of the `DataFrame` that we just parallelized, however that's not quite how Apache Spark works. Spark allows two distinct kinds of operations by the user. There are **transformations** and there are **actions**.\n",
    "\n",
    "### Transformations\n",
    "\n",
    "Transformations are operations that will not be completed at the time you write and execute the code in a cell - they will only get executed once you have called a **action**. An example of a transformation might be to convert an integer into a float or to filter a set of values.\n",
    "\n",
    "### Actions\n",
    "\n",
    "Actions are commands that are computed by Spark right at the time of their execution. They consist of running all of the previous transformations in order to get back an actual result. An action is composed of one or more jobs which consists of tasks that will be executed by the workers in parallel where possible\n",
    "\n",
    "Here are some simple examples of transformations and actions. Remember, these **are not all** the transformations and actions - this is just a short sample of them. We'll get to why Apache Spark is designed this way shortly!\n",
    "\n",
    "![transformations and actions](https://training.databricks.com/databricks_guide/gentle_introduction/trans_and_actions.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we've seen that Spark consists of actions and transformations. Let's talk about why that's the case. The reason for this is that it gives a simple way to optimize the entire pipeline of computations as opposed to the individual pieces. This makes it exceptionally fast for certain types of computation because it can perform all relevant computations at once. Technically speaking, Spark `pipelines` this computation which we can see in the image below. This means that certain computations can all be performed at once (like a map and a filter) rather than having to do one operation for all pieces of data then the following operation.\n",
    "\n",
    "![transformations and actions](https://training.databricks.com/databricks_guide/gentle_introduction/pipeline.png)\n",
    "\n",
    "Apache Spark can also keep results in memory as opposed to other frameworks that immediately write to disk after each task.\n",
    "\n",
    "## Apache Spark Architecture\n",
    "\n",
    "Before proceeding with our example, let's see an overview of the Apache Spark architecture. As mentioned before, Apache Spark allows you to treat many machines as one machine and this is done via a master-worker type architecture where there is a `driver` or master node in the cluster, accompanied by `worker` nodes. The master sends work to the workers and either instructs them to pull to data from memory or from disk (or from another data source).\n",
    "\n",
    "The diagram below shows an example Apache Spark cluster, basically there exists a Driver node that communicates with executor nodes. Each of these executor nodes have slots which are logically like execution cores. \n",
    "\n",
    "![spark-architecture](https://training.databricks.com/databricks_guide/gentle_introduction/videoss_logo.png)\n",
    "\n",
    "The Driver sends Tasks to the empty slots on the Executors when work has to be done:\n",
    "\n",
    "![spark-architecture](https://training.databricks.com/databricks_guide/gentle_introduction/spark_cluster_tasks.png)\n",
    "\n",
    "Note: In the case of the Community Edition there is no Worker, and the Master, not shown in the figure, executes the entire code.\n",
    "\n",
    "![spark-architecture](https://docs.azuredatabricks.net/_static/images/notebooks/notebook-microcluster-agnostic.png)\n",
    "\n",
    "\n",
    "## A Worked Example of Transformations and Actions\n",
    "\n",
    "To illustrate all of these architectural and most relevantly **transformations** and **actions** - let's go through a more thorough example, this time using `DataFrames` and a csv file. \n",
    "\n",
    "The DataFrame and SparkSQL work almost exactly as we have described above, we're going to build up a plan for how we're going to access the data and then finally execute that plan with an action. We'll see this process in the diagram below. We go through a process of analyzing the query, building up a plan, comparing them and then finally executing it.\n",
    "\n",
    "![Spark Query Plan](https://training.databricks.com/databricks_guide/gentle_introduction/query-plan-generation.png)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import datetime\n",
    "\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2020-01-14 16:03:54--  https://raw.githubusercontent.com/tidyverse/ggplot2/master/data-raw/diamonds.csv\n",
      "Resolving raw.githubusercontent.com... 151.101.36.133\n",
      "Connecting to raw.githubusercontent.com|151.101.36.133|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 2772143 (2.6M) [text/plain]\n",
      "Saving to: 'diamonds.csv.1'\n",
      "\n",
      "diamonds.csv.1      100%[===================>]   2.64M  --.-KB/s    in 0.1s    \n",
      "\n",
      "2020-01-14 16:03:54 (22.8 MB/s) - 'diamonds.csv.1' saved [2772143/2772143]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!wget https://raw.githubusercontent.com/tidyverse/ggplot2/master/data-raw/diamonds.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataPath = \"diamonds.csv\"\n",
    "diamonds = spark.read.format(\"csv\")\\\n",
    "  .option(\"header\",\"true\")\\\n",
    "  .option(\"inferSchema\", \"true\")\\\n",
    "  .load(dataPath)\n",
    "  \n",
    "# inferSchema means we will automatically figure out column types \n",
    "# at a cost of reading the data more than once"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we've loaded in the data, we're going to perform computations on it. This provide us a convenient tour of some of the basic functionality and some of the nice features that makes running Spark the simplest! In order to be able to perform our computations, we need to understand more about the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(carat=0.23, cut='Ideal', color='E', clarity='SI2', depth=61.5, table=55.0, price=326, x=3.95, y=3.98, z=2.43),\n",
       " Row(carat=0.21, cut='Premium', color='E', clarity='SI1', depth=59.8, table=61.0, price=326, x=3.89, y=3.84, z=2.31),\n",
       " Row(carat=0.23, cut='Good', color='E', clarity='VS1', depth=56.9, table=65.0, price=327, x=4.05, y=4.07, z=2.31),\n",
       " Row(carat=0.29, cut='Premium', color='I', clarity='VS2', depth=62.4, table=58.0, price=334, x=4.2, y=4.23, z=2.63)]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "diamonds.head(4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we've explored the data, let's return to understanding **transformations** and **actions**. I'm going to create several transformations and then an action. After that we will inspect exactly what's happening under the hood.\n",
    "\n",
    "These transformations are simple, first we group by two variables, cut and color and then compute the average price. Then we're going to inner join that to the original dataset on the column `color`. Then we'll select the average price as well as the carat from that new dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df1 = diamonds.groupBy(\"cut\", \"color\").avg(\"price\") # a simple grouping\n",
    "\n",
    "df2 = df1\\\n",
    "  .join(diamonds, on='color', how='inner')\\\n",
    "  .select(\"`avg(price)`\", \"carat\")\n",
    "# a simple join and selecting some columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These transformations are now complete in a sense but nothing has happened. As you'll see above we don't get any results back! \n",
    "\n",
    "The reason for that is these computations are *lazy* in order to build up the entire flow of data from start to finish required by the user. This is a intelligent optimization for two key reasons. Any calculation can be recomputed from the very source data allowing Apache Spark to handle any failures that occur along the way, successfully handle stragglers. Secondly, Apache Spark can optimize computation so that data and computation can be `pipelined` as we mentioned above. Therefore, with each transformation Apache Spark creates a plan for how it will perform this work.\n",
    "\n",
    "To get a sense for what this plan consists of, we can use the `explain` method. Remember that none of our computations have been executed yet, so all this explain method does is tells us the lineage for how to compute this exact dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(3) Project [avg(price)#63, carat#67]\n",
      "+- *(3) BroadcastHashJoin [color#12], [color#69], Inner, BuildRight\n",
      "   :- *(3) HashAggregate(keys=[cut#11, color#12], functions=[avg(cast(price#16 as bigint))])\n",
      "   :  +- Exchange hashpartitioning(cut#11, color#12, 200)\n",
      "   :     +- *(1) HashAggregate(keys=[cut#11, color#12], functions=[partial_avg(cast(price#16 as bigint))])\n",
      "   :        +- *(1) Project [cut#11, color#12, price#16]\n",
      "   :           +- *(1) Filter isnotnull(color#12)\n",
      "   :              +- *(1) FileScan csv [cut#11,color#12,price#16] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/data/home/mma/notebooks/diamonds.csv], PartitionFilters: [], PushedFilters: [IsNotNull(color)], ReadSchema: struct<cut:string,color:string,price:int>\n",
      "   +- BroadcastExchange HashedRelationBroadcastMode(List(input[1, string, true]))\n",
      "      +- *(2) Project [carat#67, color#69]\n",
      "         +- *(2) Filter isnotnull(color#69)\n",
      "            +- *(2) FileScan csv [carat#67,color#69] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/data/home/mma/notebooks/diamonds.csv], PartitionFilters: [], PushedFilters: [IsNotNull(color)], ReadSchema: struct<carat:double,color:string>\n"
     ]
    }
   ],
   "source": [
    "df2.explain()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now explaining the above results is outside of this introductory tutorial, but please feel free to read through it. What you should deduce from this is that Spark has generated a plan for how it hopes to execute the given query. Let's now run an action in order to execute the above plan."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "269700"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df2.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This will execute the plan that Apache Spark built up previously. Click the little arrow next to where it says `(2) Spark Jobs` after that cell finishes executing and then click the `View` link. This brings up the Apache Spark Web UI right inside of your notebook. This can also be accessed from the cluster attach button at the top of this notebook. In the Spark UI, you should see something that includes a diagram something like this.\n",
    "\n",
    "![img](https://training.databricks.com/databricks_guide/gentle_introduction/spark-dag-ui-before-2-0.png)\n",
    "\n",
    "or\n",
    "\n",
    "![img](https://training.databricks.com/databricks_guide/gentle_introduction/spark-dag-ui.png)\n",
    "\n",
    "These are significant visualizations. The top one is using Apache Spark 1.6 while the lower one is using Apache Spark 2.0, we'll be focusing on the 2.0 version. These are Directed Acyclic Graphs (DAG)s of all the computations that have to be performed in order to get to that result. It's easy to see that the second DAG visualization is much cleaner than the one before but both visualizations show us all the steps that Spark has to get our data into the final form. \n",
    "\n",
    "Again, this DAG is generated because transformations are *lazy* - while generating this series of steps Spark will optimize lots of things along the way and will even generate code to do so. This is one of the core reasons that users should be focusing on using DataFrames and Datasets instead of the legacy RDD API. With DataFrames and Datasets, Apache Spark will work under the hood to optimize the entire query plan and pipeline entire steps together. \n",
    "\n",
    "In this diagram you can see that we start with a CSV all the way on the left side, perform some changes, merge it with another CSV file (that we created from the original DataFrame), then join those together and finally perform some aggregations until we get our final result!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Caching\n",
    "\n",
    "One of the significant parts of Apache Spark is its ability to store things in memory during computation. This is a neat trick that you can use as a way to speed up access to commonly queried tables or pieces of data. This is also great for iterative algorithms that work over and over again on the same data. While many see this as a panacea for all speed issues, think of it much more like a tool that you can use. Other important concepts like data partitioning, clustering and bucketing can end up having a much greater effect on the execution of your job than caching however remember - these are all tools in your tool kit!\n",
    "\n",
    "To cache a DataFrame or RDD, simply use the cache method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[avg(price): double, carat: double]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df2.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Caching, like a transformation, is performed lazily. That means that it won't store the data in memory until you call an action on that dataset. \n",
    "\n",
    "Here's a simple example. We've created our df2 DataFrame which is essentially a logical plan that tells us how to compute that exact DataFrame. We've told Apache Spark to cache that data after we compute it for the first time. So let's call a full scan of the data with a count twice. The first time, this will create the DataFrame, cache it in memory, then return the result. The second time, rather than recomputing that whole DataFrame, it will just hit the version that it has in memory.\n",
    "\n",
    "Let's take a look at how we can discover this."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "269700"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df2.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "However after we've now counted the data. We'll see that the explain ends up being quite different."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "269700"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df2.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the above example, we can see that this cuts down on the time needed to generate this data immensely - often by at least an order of magnitude. With much larger and more complex data analysis, the gains that we get from caching can be even greater!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "## REGISTER DATA-FRAME AS A TEMP-TABLE IN SQL-CONTEXT\n",
    "diamonds.createOrReplaceTempView(\"diamonds1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "sqlStatement = \"\"\"\n",
    "    select color, count(*) as cnt\n",
    "    from diamonds1\n",
    "    where carat > 0.75\n",
    "    group by color\n",
    "    order by color\n",
    "    \n",
    "\"\"\"\n",
    "sqlResultsPD = spark.sql(sqlStatement).sample(False, 0.5, seed=1234).toPandas();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>color</th>\n",
       "      <th>cnt</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>F</td>\n",
       "      <td>3826</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>G</td>\n",
       "      <td>4869</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>H</td>\n",
       "      <td>4610</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  color   cnt\n",
       "0     F  3826\n",
       "1     G  4869\n",
       "2     H  4610"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sqlResultsPD"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 Spark - local",
   "language": "python",
   "name": "spark-3-python"
  },
  "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.5.6"
  },
  "name": "1. Introduction to Apache Spark on Databricks",
  "notebookId": 2100880988922229
 },
 "nbformat": 4,
 "nbformat_minor": 1
}