msan694_disco/MSAN694-Lec5-Notes.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## MSAN-694-Lec5-Notes\n",
    "\n",
    "Gonna have some fun with AWS\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Week 4 Material: Recapping The Pair RDD\n",
    "- countByKey()\n",
    "- lookupKey()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "PATH2 = '/Users/tlee010/Desktop/github_repos/2017-msan694-example/Data/filtered_registered_business_sf.csv'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From ”filtered_registered_business_sf.csv”, create a pair RDD of (zip, (store name, city))\n",
    "- Countpairswhichdonothaveakey.\n",
    "- Filterpairsthatdonotinclude“San Francisco” in the city value."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Count Pairs which don't have a key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "92"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines = sc.textFile(PATH2)\n",
    "lines.collect()\n",
    "\n",
    "parse = lines.map(lambda x : x.split(','))\n",
    "parse = parse.map(lambda x : (x[0],(x[1],x[2])))\n",
    "\n",
    "len(parse.lookup(''))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "92"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "parse.countByKey()['']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Filter pairs that do not include “San Francisco” in the city value."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "198561"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines = sc.textFile(PATH2)\n",
    "lines.collect()\n",
    "\n",
    "parse = lines.map(lambda x : x.split(','))\n",
    "parse = parse.map(lambda x : (x[0],(x[1],x[2])))\n",
    "parse.filter(lambda x : x[1][1]).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Week 4: About them RDDs (resilient distributed dataset)\n",
    "\n",
    "### 4.1 - how to add some data redundancy for data/system failures\n",
    "\n",
    "RDDs are by default **recomputed each time**. calls:\n",
    "\n",
    "---\n",
    "\n",
    "To cache the data for repeat:\n",
    " \n",
    "**`line_with_spark.persist(StorageLevel.persistency_level)`**\n",
    "\n",
    "**`persistency_level`** - storage levels can be the following:\n",
    "\n",
    "- **`MEMORY_ONLY`** : put as much possible in RAM, recompute the rest\n",
    "\n",
    "- **`MEMORY_AND_DISK`** : put as much as possible in MEM, remainder on DISK\n",
    "\n",
    "- **`MEMORY_ONLY_SER`** : turn data into a stream of bytes ( reduction AKA serialized) but needs to conver the data back later\n",
    "\n",
    "- **`MEMORY_AND_DISK_ONLY_SER`** : turn data into a stream of bytes ( reduction AKA serialized) but needs to conver the da\"ta back later\n",
    "    \n",
    "- **`DISK_ONLY`** - keeps RDD data only on disk\n",
    "    \n",
    "- **`MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc`** - replicate each partition on two different nodes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "### Why Replicate?\n",
    "\n",
    "If a server fails, then you will have much faster recovery during computation. And for repeat calls, "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example\n",
    "\n",
    "Python example \n",
    "```python\n",
    "             .persist(storage_label)\n",
    "    .cache()=.persist(MEMORY_ONLY)\n",
    "             .persist()\n",
    "             \n",
    "             .unpersist() # restore the data\n",
    "```\n",
    "\n",
    "**`.cache()`** is the same thing as **`.persist(MEMORY_ONLY)`**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 6 - comparison\n",
    "\n",
    "\n",
    "#### No persist management\n",
    "```python\n",
    "start = time.time()\n",
    "print \"first \",\n",
    "print time.time() - start\n",
    "\n",
    "start = time.time()\n",
    "lines.count()\n",
    "print \"second \",\n",
    "print time.time() - start\n",
    "```\n",
    "#### Check the time\n",
    "```{r}\n",
    "first 59.7 secs\n",
    "second 56l72 secs\n",
    "```\n",
    "\n",
    "#### No persist management\n",
    "```python\n",
    "lines.persist(StorageLevel.MEMORY_AND_DISK)\n",
    "start = time.time()\n",
    "print \"first \",\n",
    "print time.time() - start\n",
    "\n",
    "start = time.time()\n",
    "lines.count()\n",
    "print \"second \",\n",
    "print time.time() - start\n",
    "\n",
    "start = time.time()\n",
    "lines.count()\n",
    "print \"third \",\n",
    "print time.time() - start\n",
    "```\n",
    "#### Check the time, repeat times have faster processing\n",
    "```{r}\n",
    "first 79.07\n",
    "second 20.86\n",
    "third 20.4\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark Cluster\n",
    "\n",
    "- Hadoop YARN\n",
    "- Mesos\n",
    "- Spark standalone\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get storage level"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**`getStorageLevel()`** – returns different storage option flags set for an RDD.`\n",
    "\n",
    "- **`StorageLevel(useDisk, useMemory, useOffHeap, deserialized, replication = 1)`** \n",
    "    - **`useDisk`** : If set, partitions that do not fit in memory will be written to disk.\n",
    "    - **`useMemory`** : If set, the RDDs will be stored in-memory.\n",
    "    - **`useOffHeap`** : If set, the RDD will be stored outside of the Spark executor in an external system such as Tachyon.\n",
    "    - **`deserialization`** : If set, the RDD will be stored as deserialized Java objects.\n",
    "    - **`replication`** : An integer that controls the number of copies of the persisted data to be stored."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Storage Guidance when designing a Spark Program\n",
    "\n",
    "- **`'MEMORY_ONLY'`** - Best\n",
    "- **`'DISK'`** - only if recomputation is expensive and cannot fit in memory\n",
    "- **Off-Heap** - stored outside of the spark cluster in an external system. Use if there are memory issues\n",
    "- **serialization** - if there is memory issues, or if its too big to fit into memory\n",
    "- **replication** - faster fault recovery, but the trade off is taking more storage space on your cluster. Use when there is a bad connection to your cluster, or you have live web app"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sample:\n",
    "\n",
    "```python\n",
    "    Output[1]:\n",
    "        StorageLevel(False,False,False,1)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "PATH2 = '/Users/tlee010/Desktop/github_repos/2017-msan694-example/Data/USF_Mission.txt'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.storagelevel import StorageLevel\n",
    "lines = sc.textFile(PATH2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 7\n",
    "\n",
    "Try persisting some of the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "/Users/tlee010/Desktop/github_repos/2017-msan694-example/Data/USF_Mission.txt MapPartitionsRDD[35] at textFile at NativeMethodAccessorImpl.java:0"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines.persist(StorageLevel.DISK_ONLY)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "24"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# will error, no java\n",
    "#lines.persist(StorageLevel.MEMORY_AND_DISK_SER)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# will error\n",
    "#lines.persist(StorageLevel.MEMORY_ONLY_2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### One an action has been performed, the persisted data will show in the `storage` tab"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "![](http://apache-spark-user-list.1001560.n3.nabble.com/attachment/6487/0/Cloudera-Training-Spark-Developer-VM-cdh5.0_ALPHA.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Week 5 Material : Running Spark on a Cluster\n",
    "\n",
    "### Spark Cluster\n",
    "\n",
    "### Spark Standalone Cluster - using spark instead of Mesos / YARN\n",
    "\n",
    "### Spark Cluster Types:\n",
    "- Spark Standalone\n",
    "- Hadoop YARN\n",
    "- Mesos"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Components (when running)\n",
    "\n",
    "![](https://jaceklaskowski.gitbooks.io/mastering-apache-spark/images/driver-sparkcontext-clustermanager-workers-executors.png)\n",
    "\n",
    "**Client** \n",
    "- starts the driver program.\n",
    "- prepares config + class options\n",
    "- `spark-submit`, `pyspark`, or `spark-shell` scripts\n",
    "\n",
    "**Driver** \n",
    "- 1 driver per application\n",
    "- Monitors executor of spark\n",
    "- distributes tasks to each work executor\n",
    "\n",
    "**Executor** \n",
    "- stores spark ask with variable amount of cores\n",
    "- stores and caches data in memory\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploy Mode\n",
    "\n",
    "#### Example: Driver in the cluser\n",
    "![](https://cdn-images-1.medium.com/max/568/1*-d0reAUnrkZQhL72ks8paA.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "#### Example: Driver outside the cluster\n",
    "![](http://freecontent.manning.com/wp-content/uploads/bonaci_runtimeArch_02.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark Cluster \n",
    "\n",
    "\n",
    "![](https://i.stack.imgur.com/KHekf.png)\n",
    "\n",
    "**Cluster Manager **\n",
    "- monitors worker and allocates and reserves resources (processing, storage, memory). Kinda like my old boss at a restaurant\n",
    "\n",
    "**Master**\n",
    "- Takes in applications\n",
    "- Requests resources\n",
    "- Sometimes acts as a resource manager\n",
    "\n",
    "**Worker**\n",
    "- Instructions from master\n",
    "- launches executors on worker\n",
    "- Spark must be on every node\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Good Linkedin Article\n",
    "\n",
    "https://www.linkedin.com/pulse/spark-architecture-ashish-kumar/\n",
    "\n",
    "![](https://media.licdn.com/mpr/mpr/AAEAAQAAAAAAAAs-AAAAJGIzNmIyOWFmLTVkODEtNDBlYS1hNGE4LTI1ZGE2NTQyZTQzYg.jpg)\n",
    "\n",
    "Spark Application Workflow in Standalone Mode\n",
    "\n",
    "1.     Client connect to master.\n",
    "\n",
    "2.     Master start driver on one of node.\n",
    "\n",
    "3.     Driver connect to master and request for Executors to run the tasks.\n",
    "\n",
    "4.     Master connect to worker node and request to create executors.\n",
    "\n",
    "5.     Each Worker node create one executor for each Application.\n",
    "\n",
    "6.     Driver connect to executors and schedule tasks on it.\n",
    "\n",
    "7.     Update the status of task to driver.\n",
    "\n",
    "8.     Driver send application output to client.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}