W3-02a-Pyspark-Tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark Tutorial\n",
    "<div>\n",
    " <h2> CSCI 4283 / 5253 \n",
    "  <IMG SRC=\"https://www.colorado.edu/cs/profiles/express/themes/cuspirit/logo.png\" WIDTH=50 ALIGN=\"right\"/> </h2>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark was originally developed using Scala, although there are Python and Java interfaces as well. This tutorial covers [most of the RDD API](https://spark.apache.org/docs/latest/rdd-programming-guide.html#resilient-distributed-datasets-rdds) using Python bindings.\n",
    "\n",
    "You may want to consult the [PySpark manual](http://spark.apache.org/docs/2.1.0/api/python/pyspark.html) as well."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark import SparkContext, SparkConf\n",
    "import numpy as np\n",
    "import operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `SparkContext` object tells Spark how to access a cluster. The `SparkConf` object defines information about our job.\n",
    "\n",
    "The `master` of a Spark configuration is the cluster (YARN or Mesos) manager. It can also be \"local\" meaning that the Spark job runs on your local machine, which is what we'll do here; the `[*]` notation means to use all the available cores. In general, you shouldn't hardcode the `master` mechanism.\n",
    "\n",
    "Spark uses a function chaining notation. We'll use that throughout unless it makes this confusing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf=SparkConf().setAppName(\"pyspark tutorial\").setMaster(\"local[2]\")\n",
    "sc = SparkContext(conf=conf)\n",
    "#sc.setLogLevel(\"DEBUG\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Datasets\n",
    "The basic Spark data structure is the RDD (resilient distributed data), which is essentially a vector distributed across the cluster of nodes or on the local system. In PySpark, the vector can contain a heterogenous collection of types (strings, ints, etc).\n",
    "\n",
    "You can create an RDD from a list or tuple, read it from a local file or read it from networked distributions such as HDFS or S3.\n",
    "\n",
    "The following shows creating three datasets from lists using the _parallelize_ method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "a = sc.parallelize([7, 2, 3, 1, 2, 3, 4, 5, 6, 7])\n",
    "b = sc.parallelize([2, 3, 99, 22, -77])\n",
    "c = sc.parallelize([ (1,2), (2,3), (1, 99), (3, 44), (2, 1), (4,5), (3, 19) ] )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "passwd = sc.textFile(\"/etc/passwd\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It is also possible to read and write binary data files, including data formatted in Hadoop Sequence types.\n",
    "\n",
    "Spark also supports _accumulators_ and _broadcast variables_.  Accumulators are designed to sum or aggregate values from across the cluster; they are really only suitable for commutative-associative operators. Broadcast variables are efficiently disseminated to all nodes in the cluster; they can be used for the equivilent of \"map-side joins\"."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformations and Actions\n",
    "*Transformations* produce new RDD's by transforming existing RDD's  and *Actions* convert data *to* and *from* an RDD."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Actions\n",
    "\n",
    "Some of the most simple actions are:\n",
    "* count() - Return the number of items in the RDD\n",
    "* take(_n_) - Extract and return the first _n_ items from the RDD\n",
    "* first() - Same as take(1)\n",
    "* collect() - Same as take(count()) - **returns full RDD**\n",
    "* takeSample(_withReplacement_:Boolean, _num_:int, [ seed:Int] ) - extract a random set of _num_ items from the RDD with or without replacement.\n",
    "* takeOrdered( _num_ ) - extract _num_ items from the sorted RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "print(a.count())\n",
    "print(a.take(2))\n",
    "print(a.first())\n",
    "print(a.collect())\n",
    "print(a.takeSample(True, 3))\n",
    "print(a.takeOrdered(4))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lambda Functions in Python\n",
    "\n",
    "Lambda functions, or anonymous functions are common in other languages (e.g. Scala) and commonly used in PySpark. The Python lambda is restricted to simple single-line statements."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "(lambda x: x + x)(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lambda x: x + 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "add2 = lambda x: x + 2\n",
    "add2(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Map, Reduce & flatMap\n",
    "\n",
    "`map` is a transformation that produces a new RDD. `reduce` is an action that applies a specified function to the elements of an RDD. Map is applied using a single-argument (unary) function (often a *lambda*) while reduce takes a binary (or dyadic) function.\n",
    "\n",
    "Examples:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "a.map( lambda x: x**2 ).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**reduce** does a reduction for a function across the RDD. For example, the `operator.add` function in Python is the function that impements addition, so we can reduce an RDD using:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.reduce(operator.add)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following `lambda` sums the first and second element of the tuples in `c`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "c.reduce( lambda x,y: (x[0] + y[0], x[1] + y[1]) )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That should produce the same result as the more complex example\n",
    "below, which returns an RDD for each field of the tuple and then\n",
    "adds those those using reduce. The operator.add function is \"+\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "( c.map( lambda x : x[0] ).reduce(operator.add), \n",
    "  c.map( lambda x: x[1] ).reduce(operator.add) )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**flatMap** applies a map operation across elements of a list, but then takes those elements and *appends* them to the list. The result is useful when processing a set of tuples or breaking documents into words and then processing the words rather than lines-of-words."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "sent = sc.parallelize([\"these are some\", \"sample words\" ])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sent.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "sent.map( lambda x : x.split() ).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "sent.flatMap( lambda x : x.split() ).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Filter, Sampling & Sorting\n",
    "\n",
    "Filter can be used to remove or filter items from an RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "isEven = lambda x: x %2 == 0\n",
    "\n",
    "print(a.collect())\n",
    "print(a.filter(isEven).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "passwd.map( lambda x : x.split(':' ) )\\\n",
    "   .filter( lambda x : x[0] == 'root' )\\\n",
    "   .collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we want to get a representative sample of the data, we can use **takeSample** to sample (and collect) that data or **sample** to get an RDD. The first argument determines if we sample with replacement (true) or not."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.takeSample(True, 3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.sample(True, 3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you're examining data using **take** but you want to access the underlying data in a sorted order you can use **takeOrdered**. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.takeOrdered(4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also use **takeOrdered** with more complex pairs if you provide an ordering function. For example, we can sort (K,V) pairs using the extra function to define the order:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The default **takeOrdered** uses the sort-order of the type, which sorts on the keys (and the values if equal keys)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.takeOrdered(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can provide an ordering function as a lambda or function of a single argument. We can also sort in reverse order.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.takeOrdered(3, lambda x: x[1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.takeOrdered(3, lambda x: -x[1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**sortBy** and **sortByKey** serve a similar role as takeOrdered but completely sorts an RDD rather than the just the returned results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "passwdLines = open('/etc/passwd', 'r').readlines()\n",
    "passwd = sc.parallelize( passwdLines )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "userAndShell = passwd.map( lambda x: x.rstrip('\\n').split(':') )\\\n",
    "    .map( lambda y: ( y[0], y[6] ) )\n",
    "userAndShell.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`sortBy( cmp: Func, ascending: Boolean)` takes a function that returns the sort key."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "userAndShell.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "userAndShell.sortBy(lambda x : x[0] ).take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "userAndShell.sortBy(lambda x : x[1] ).take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`sortByKey( ascending: Boolean)` assumes the data is in (k,v) pairs. In this case, the example is the same as sortByUser above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "userAndShell.sortByKey().take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Set Operations\n",
    "\n",
    "**union** and **intersection** produce new RDD's where the elements can be thought of as being in a set. **distinct** returns the unique set of items in an RDD (*i.e.* converting a multi-set to a set). **sample**(withReplacement:Boolean, fraction:Float, [seed:int]) draws samples with or without replacement. Sample produces more representative samples with larger datasets and has seemingly erratic behavior with small sets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "a.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "b.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "a.union(b).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "a.intersection(b).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`subtract` removes items from the RDD that are contained in a second RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "print(a.collect(), \" - \", b.collect(), \" = \", a.subtract(b).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "a.distinct().collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}