module-5/day-8-pyspark/spark-intro.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Spark Machine learning \n",
    "\n",
    "### Learning goals:\n",
    "- align the relationships between Hadoop, Spark\n",
    "- differentiate between Spark RDDs and Spark Dataframes and when each is appropriate\n",
    "- locate and explore the Spark.ML documentation\n",
    "- code along a text classification problem using four different ml algorithms, a data prep pipeline, and gridsearch to fine tune a model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Spark context and concepts review\n",
    "<img src=\"https://images.pexels.com/photos/285173/pexels-photo-285173.jpeg?auto=compress&cs=tinysrgb&dpr=2&h=750&w=1260\" width=600>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# The story of Spark (in diagrams)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "## Start with Hadoop\n",
    "<img alt=\"diagram of hadoop v1 compared to hadoop v2\" src=\"img/yarn.png\" width=800>\n",
    "\n",
    "[diagram source](https://sites.google.com/site/codingbughunter/hadoop/yarn-general-discribe)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Yarn facilitates the resource allocation between Spark and the HDFS\n",
    "### YARN = Yet Another Resource Negotiator\n",
    "#### YARN is a subproduct of Hadoop\n",
    "![yarn diagram with spark](http://hortonworks.com/wp-content/uploads/2013/06/YARN.png)\n",
    "\n",
    "[diagram source](https://sites.google.com/site/codingbughunter/hadoop/yarn-general-discribe)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Then visualize the Spark ecosystem built on top of that\n",
    "\n",
    "<img alt=\"diagram of spark eco system components\" src=\"img/spark_eco.png\" width=600>\n",
    "\n",
    "[image source here](https://databricks.com/spark/about)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# The story of Spark (a timeline)\n",
    "\n",
    "|<p align=\"left justify\">Date</p>|<p align=\"left justify\">Product</p>|<p align=\"left justify\">Update</p>|\n",
    "|:----|:-----|:-----|\n",
    "| 2002 | Hadoop | <p align=\"left justify\">Doug Cutting starts `Apache Nutch` researching sort/merge processing</p> |\n",
    "| 2006 | Hadoop |  <p align=\"left justify\">Leaves `Nutch` and joins `Yahoo`, renaming the project `Hadoop` </p>|\n",
    "| 2008 | Hadoop |  <p align=\"left justify\">`Hadoop` was made `Apache’s` top level project </p> |\n",
    "| Jan 2008 | Hadoop |  <p align=\"left justify\">v 0.10.1 released </p>|\n",
    "| 2009 | Spark | <p align=\"left justify\">started as a research project at the UC Berkeley AMPLab  </p>|\n",
    "| 2010 | Spark |  <p align=\"left justify\">open sourced </p>|\n",
    "| Sept 2012 | Spark |  <p align=\"left justify\">0.6.0 released </p>|\n",
    "| 2013 | Spark |  <p align=\"left justify\">moved to the `Apache` Software Foundation </p>|\n",
    "| Feb 2013| Spark |  <p align=\"left justify\">Spark 0.7 adds a Python API called `PySpark` </p>|\n",
    "| Sept 2013 | Spark | <p align=\"left justify\">0.8.0 introduces `MLlib` </p>|\n",
    "| 2013 | Databricks |  <p align=\"left justify\">Original Spark research team at UC Berkeley found Databricks</p> |\n",
    "| May 2014 |Spark |  <p align=\"left justify\">v 1.0 introduces Spark SQL, for loading and manipulating structured data in Spark</p>|\n",
    "|Mar 2015 | Spark | <p align=\"left justify\"> v 1.3.0 brings a new DataFrame API</p> |\n",
    "| Jan 2016|  Spark | <p align=\"left justify\"> v 1.6.0 brings a new Dataset API </p> |\n",
    "| Jul 2016 | Spark | <p align=\"left justify\"> v 2.0.0 In Scala and Java, DataFrame and Dataset have been unified, i.e. DataFrame is just a type alias for Dataset of Row. In Python and R DataFrame is the main programming interface. <br> - SparkSession: new entry point that replaces the old SQLContext<br>- Native CSV data source, based on Databricks’ spark-csv module<br>- MLlib - The DataFrame-based API is now the primary API. The RDD-based API is entering maintenance mode </p> |\n",
    "| Jul 2017| Spark | <p align=\"left justify\"> v 2.2.0 drops support for Python 2.6 |\n",
    "| Nov 2018 | Spark | <p align=\"left justify\"> v 2.4.0<br> - This release adds Barrier Execution Mode for better integration with deep learning frameworks<br> - more integration between pandas UDF and spark DataFrames </p>|\n",
    "| Jan 2020| PySpark | <p align=\"left justify\"> Python 2 will be officially unsupported |\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Spark data objects\n",
    "\n",
    "<img alt=\"diagram of definitions of Spark objects from databricks\" src=\"https://databricks.com/wp-content/uploads/2018/05/rdd-1024x595.png\" width=700>"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Differences between objects:\n",
    "<img alt=\"memory usage\" src=\"https://databricks.com/wp-content/uploads/2016/07/memory-usage-when-caching-datasets-vs-rdds.png\" width=600>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img alt=\"type safety\" src=\"https://databricks.com/wp-content/uploads/2016/07/sql-vs-dataframes-vs-datasets-type-safety-spectrum.png\" width=600>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Use an RDD when:\n",
    "[quoted from databricks](https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html)\n",
    "\n",
    "> - you want low-level transformation and actions and control on your dataset;\n",
    "> - your data is unstructured, such as media streams or streams of text;\n",
    "> - you want to manipulate your data with functional programming constructs than domain specific expressions;\n",
    "> - you don’t care about imposing a schema, such as columnar format, while processing or accessing data attributes by name or column"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Use a dataframe when:\n",
    "[also quoted from databricks](https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html)\n",
    "\n",
    "\n",
    "> - you want rich semantics, high-level abstractions, and domain specific APIs, use DataFrame\n",
    "> - your processing demands high-level expressions, filters, maps, aggregation, averages, sum, SQL queries, columnar access and use of lambda functions on semi-structured data, use DataFrame\n",
    "> - you want higher degree of type-safety at compile time, want typed JVM objects, take advantage of Catalyst optimization, and benefit from Tungsten’s efficient code generation, use Dataset.\n",
    "> - you want unification and simplification of APIs across Spark Libraries, use DataFrame or Dataset.\n",
    "> - If you are a R user, use DataFrames.\n",
    "> - If you are a Python user, use DataFrames and resort back to RDDs if you need more control.\n",
    "\n",
    "**Note**: Machine learning algorithms are run on _DataFrames_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Review:\n",
    "\n",
    "- You are grabbing live tweets about the CW show 'Jane the Virgin' for later analysis. In the Spark ecosystem, where should you store them? an RDD or a DataFrame?\n",
    "\n",
    "- You have an RDD of data that you wish to use to build a predictive model. Should you leave it as an RDD or transform it to a DataFrame?\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 1: Create spark context"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "spark = pyspark.sql.SparkSession.builder.getOrCreate()\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Create a pandas dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!head data/users.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "p_df = pd.read_csv('data/users.csv', sep='|') "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Create spark dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd = sc.textFile(\"data/users.csv\").map(lambda line: line.split(\"|\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(\"data/users.csv\", sep=\"|\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# alternatively:\n",
    "from_pandas_df = spark.createDataFrame(p_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# it's already a DF, but this is the easy way to rename columns\n",
    "df = (spark.read.csv(\"data/users.csv\", sep=\"|\")\n",
    "           .toDF(\"id\", \"age\", \"gender\", \"occupation\", \"zip\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "(\n",
    "    df.where(\"occupation != 'other'\")\n",
    "      .groupby(\"occupation\")\n",
    "      .count()\n",
    "      .sort(\"count\", ascending=0)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df.persist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# df.limit(5).toPandas()\n",
    "# df.head()\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as F\n",
    "df.agg(F.countDistinct('occupation')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We can also use SQL to query\n",
    "df.createOrReplaceTempView('users')\n",
    "\n",
    "query = \"\"\"\n",
    "SELECT occupation, COUNT(*) as count\n",
    "FROM users\n",
    "GROUP BY occupation\n",
    "ORDER BY count DESC\n",
    "\"\"\"\n",
    "output = spark.sql(query)\n",
    "output.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}