Lab 1 - Introduction to Spark and HPC.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# COM6012 Scalable Machine Learning 2019 - Haiping Lu\n",
    "# Lab 1 - Introduction to (Py)Spark and (Sheffield)HPC\n",
    "\n",
    "## Objectives\n",
    "\n",
    "* Task 1: To finish in the lab session. **Critical**\n",
    "* Task 2: To finish in the lab session. **Critical**\n",
    "* Task 3: To finish in the lab session. **Essential**\n",
    "* Task 4: To finish in the lab session. **Essential**\n",
    "* Task 5: To explore by yourself before the next session. **Optional but recommended**\n",
    "\n",
    "**Suggested reading**: \n",
    "* Chapters 2 to 4 of [PySpark tutorial](https://runawayhorse001.github.io/LearningApacheSpark/pyspark.pdf) (several sections in Chapter 3 can be safely skipped)\n",
    "* [Spark Quick Start](https://spark.apache.org/docs/2.3.2/quick-start.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "All slides and notebooks are at: https://github.com/haipinglu/ScalableML/\n",
    "\n",
    "If you'd like to fork it, here is a guide on how to update it: see **[How do I update a GitHub forked repository?](https://stackoverflow.com/questions/7244321/how-do-i-update-a-github-forked-repository)**\n",
    "\n",
    "**Note**: We will use pyspark **2.3.2** rather than the latest **2.4.0** due to JVM problem on Windows. See this [discussion](https://stackoverflow.com/questions/53161939/pyspark-error-does-not-exist-in-the-jvm-error-when-initializing-sparkcontext)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Install Spark\n",
    "\n",
    "### 1.1. HPC - ShARC  (help: ` hpc@sheffield.ac.uk`)\n",
    "\n",
    "#### Connect to ShARC (`host: sharc.sheffield.ac.uk`) via SSH: [MobaXterm on Windows](https://www.sheffield.ac.uk/cics/research/hpc/using/access/windowspc),  [Linux/Unix](https://www.sheffield.ac.uk/cics/research/hpc/using/access/linux), [Apple/Mac](https://www.sheffield.ac.uk/cics/research/hpc/using/access/apple). See [Connecting to a cluster using SSH](http://docs.hpc.shef.ac.uk/en/latest/hpc/connecting.html), and also [Intro_to_HPC by Mike](https://github.com/mikecroucher/Intro_to_HPC) (note that we are not using Scala though)\n",
    "\n",
    "#### Start an interactive session on a node\n",
    "`qrshx`\n",
    "#### Load java and conda\n",
    "`module load apps/java/jdk1.8.0_102/binary`\n",
    "\n",
    "`module load apps/python/conda`\n",
    "\n",
    "#### Create a virtual environment called myspark\n",
    "`conda create -n myspark python=3.6`\n",
    "\n",
    "#### Activate the environment\n",
    "`source activate myspark`\n",
    "\n",
    "#### Install spark and pyspark 2.3.2 from conda-forge\n",
    "`conda install -c conda-forge pyspark=2.3.2`\n",
    "\n",
    "#### Run spark\n",
    "`pyspark`\n",
    "\n",
    "#### Not familiar with terminal?: [A tutorial from Mike Croucher](https://github.com/mikecroucher/Intro_to_HPC/blob/gh-pages/terminal_tutorial.md)\n",
    "\n",
    "### 1.2. Windows/Linux/Mac: on your own / lab machine.  \n",
    "\n",
    "You need to follow instructions for your OS (Windows/Linux/Mac) below to install Java, set up the proper paths, etc., except that if you have `conda` installed already (e.g., from COM6509 the MLAI module), `pyspark 2.3.2` can be installed via (see above)\n",
    "\n",
    "`conda install -c conda-forge pyspark=2.3.2`\n",
    "\n",
    "\n",
    "* Windows: 1) With video - [Install Spark on Windows (PySpark)](https://medium.com/@GalarnykMichael/install-spark-on-windows-pyspark-4498a5d8d66c) 2) [How to install Spark on Windows in 5 steps](https://medium.com/@dvainrub/how-to-install-apache-spark-2-x-in-your-pc-e2047246ffc3) **Note:** The following may be needed. Go to your System Environment Variables and add PYTHONPATH to it with the following value: `%SPARK_HOME%\\python;%SPARK_HOME%\\python\\lib\\py4j-<version>-src.zip:%PYTHONPATH%`, just check what py4j version you have in your `spark/python/lib` folder ([source](https://stackoverflow.com/questions/53161939/pyspark-error-does-not-exist-in-the-jvm-error-when-initializing-sparkcontext?noredirect=1&lq=1)).\n",
    "\n",
    "* Linux: With video - [Install PySpark on Ubuntu](https://medium.com/@GalarnykMichael/install-spark-on-ubuntu-pyspark-231c45677de0)\n",
    "\n",
    "* Mac: [Install Spark/PySpark on Mac](https://medium.com/@yajieli/installing-spark-pyspark-on-mac-and-fix-of-some-common-errors-355a9050f735) **Note: you need to use Java 8. Java 11 is having problems.**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Quiz Pyspark shell by `Ctrl+D`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Run Spark\n",
    "\n",
    "### On HPC: connect and activate first\n",
    "\n",
    "`qrshx`\n",
    "\n",
    "`module load apps/java/jdk1.8.0_102/binary`\n",
    "\n",
    "`module load apps/python/conda`\n",
    "\n",
    "`source activate myspark`\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Interactive (HPC or local machine)\n",
    "\n",
    "#### If running notebook of spark on your local machine\n",
    "**Note** If `import pyspark` reports error, you may try `pip install findspark`, `import findspark`, \n",
    "`findspark.init()`, and then `import pyspark` should work."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import findspark\n",
    "#findspark.init()\n",
    "import pyspark\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[2]\") \\\n",
    "    .appName(\"COM6012 Spark Intro\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### If running spark in a shell on either HPC or your local machine, `spark` (SparkSession) and `sc` (SparkContext) is automatically created.\n",
    "\n",
    "Run pyspark (optionally, specify to use multiple cores)\n",
    "\n",
    "`pyspark` or `pyspark --master local[2]` with two cores\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check your SparkSession and SparkContext object (you will see different output if running in a shell)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://143.167.111.234:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.3.2</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[2]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>COM6012 Spark Intro</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x1e114963d68>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create and check sc (SparkContext)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://143.167.111.234:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.3.2</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[2]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>COM6012 Spark Intro</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[2] appName=COM6012 Spark Intro>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1, 4, 9, 16]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "nums = sc.parallelize([1,2,3,4])\n",
    "nums.map(lambda x: x*x).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Log Mining with Spark - Example\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This example deals with **Semi-Structured** data in a text file. \n",
    "\n",
    "Firstly, you need to **make sure the file is in the proper directory and change the file path if necessary**, on either HPC or local machine.\n",
    "\n",
    "**If running on HPC, you need to transfer files there.** Here is how to [**transfer files to HPC**](https://www.sheffield.ac.uk/cics/research/hpc/using/access). Please **click** and follow the instructions unless you are already familiar with it.\n",
    "\n",
    "### GUI-based file transfer\n",
    "\n",
    "* [**MobaXterm**](https://mobaxterm.mobatek.net/) is recommended for **Windows**\n",
    "* [**Cyberduck**](https://en.wikipedia.org/wiki/Cyberduck) or [**FileZilla**](https://en.wikipedia.org/wiki/FileZilla) is recommended for **Mac**\n",
    "* **FileZilla** is recommended for **Linux (e.g., Ubuntu)**\n",
    "\n",
    "For example, in MobaXterm (for Windows), you just need to **Drag your file or folder to the left directory pane of MobaXterm**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[value: string]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "logFile=spark.read.text(\"Data/NASA_Aug95_100.txt\")\n",
    "logFile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "100"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "logFile.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(value='in24.inetnebr.com - - [01/Aug/1995:00:00:01 -0400] \"GET /shuttle/missions/sts-68/news/sts-68-mcc-05.txt HTTP/1.0\" 200 1839')"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "logFile.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Question**: How many accesses are from Japan?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "hostsJapan = logFile.filter(logFile.value.contains(\".jp\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check whether you are getting what you want."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------------------------------------------------------------------------------------+\n",
      "|value                                                                                                         |\n",
      "+--------------------------------------------------------------------------------------------------------------+\n",
      "|kgtyk4.kj.yamagata-u.ac.jp - - [01/Aug/1995:00:00:17 -0400] \"GET / HTTP/1.0\" 200 7280                         |\n",
      "|kgtyk4.kj.yamagata-u.ac.jp - - [01/Aug/1995:00:00:18 -0400] \"GET /images/ksclogo-medium.gif HTTP/1.0\" 200 5866|\n",
      "|kgtyk4.kj.yamagata-u.ac.jp - - [01/Aug/1995:00:00:21 -0400] \"GET /images/NASA-logosmall.gif HTTP/1.0\" 304 0   |\n",
      "|kgtyk4.kj.yamagata-u.ac.jp - - [01/Aug/1995:00:00:21 -0400] \"GET /images/MOSAIC-logosmall.gif HTTP/1.0\" 304 0 |\n",
      "|kgtyk4.kj.yamagata-u.ac.jp - - [01/Aug/1995:00:00:22 -0400] \"GET /images/USA-logosmall.gif HTTP/1.0\" 304 0    |\n",
      "+--------------------------------------------------------------------------------------------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "hostsJapan.show(5,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "hostsJapan.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Self-contained Application\n",
    "\n",
    "To run a self-contained application, you need to **exit your shell, by `Ctrl+D` first**.\n",
    "\n",
    "Create a file `LogMining100.py`\n",
    "\n",
    "~~~~\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[2]\") \\\n",
    "    .appName(\"COM6012 Spark Intro\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext\n",
    "sc.setLogLevel(\"WARN\")\n",
    "\n",
    "logFile=spark.read.text(\"Data/NASA_Aug95_100.txt\")\n",
    "hostsJapan = logFile.filter(logFile.value.contains(\".jp\")).count()\n",
    "\n",
    "print(\"\\n\\nHello Spark: There are %i hosts from Japan.\\n\\n\" % (hostsJapan))\n",
    "\n",
    "spark.stop()\n",
    "~~~~\n",
    "\n",
    "\n",
    "Then run it with `spark-submit Code/LogMining100.py`  **Note: You need exit your shell, by `Ctrl+D` first**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "## 4. Big Data Log Mining with Spark \n",
    "\n",
    "**Data**: Download the August data in gzip (NASA_access_log_Aug95.gz) from [NASA HTTP server access log](http://ita.ee.lbl.gov/html/contrib/NASA-HTTP.html) and put into your `Data` folder. `NASA_Aug95_100.txt` above is the first 100 lines of the August data.\n",
    "\n",
    "**Question**: How many accesses are from Japan and UK respectively?\n",
    "\n",
    "Create a file `LogMiningBig.py`\n",
    "\n",
    "~~~~\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[2]\") \\\n",
    "    .appName(\"COM6012 Spark Intro\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext\n",
    "sc.setLogLevel(\"WARN\")\n",
    "\n",
    "logFile=spark.read.text(\"../Data/NASA_access_log_Aug95.gz\").cache()\n",
    "\n",
    "hostsJapan = logFile.filter(logFile.value.contains(\".jp\")).count()\n",
    "hostsUK = logFile.filter(logFile.value.contains(\".uk\")).count()\n",
    "\n",
    "print(\"\\n\\nHello Spark: There are %i hosts from UK.\\n\" % (hostsUK))\n",
    "print(\"Hello Spark: There are %i hosts from Japan.\\n\\n\" % (hostsJapan))\n",
    "\n",
    "spark.stop()\n",
    "~~~~\n",
    "**Spark can read gzip file directly. You do not need to unzip it to a big file.**\n",
    "\n",
    "**Note the use of cache() above**\n",
    "\n",
    "### Run a program in batch mode\n",
    "\n",
    "[How to submi batch jobs to ShARC](https://www.sheffield.ac.uk/cics/research/hpc/sharc/batch) **The more resources you request, the longer you need to queue**\n",
    "\n",
    "Interactive mode will be good for learning, exploring and debugging, with smaller data. For big data, it will be more convenient to use batch processing. You submit the job to the node to join a queue. Once allocated, your job will run, with output properly recorded. This is done via a shell script.\n",
    "\n",
    "Create a file `Lab1_SubmitBatch.sh`\n",
    "\n",
    "~~~~\n",
    "#!/bin/bash\n",
    "#$ -l h_rt=2:00:00  #time needed\n",
    "#$ -pe smp 2 #number of cores\n",
    "#$ -l rmem=4G #number of memery\n",
    "#$ -o COM6012_Lab1.output #This is where your output and errors are logged.\n",
    "#$ -j y # normal and error outputs into a single file (the file above)\n",
    "#$ -M youremail@shef.ac.uk #Notify you by email, remove this line if you don't like\n",
    "#$ -m ea #Email you when it finished or aborted\n",
    "#$ -cwd # Run job from current directory\n",
    "\n",
    "module load apps/java/jdk1.8.0_102/binary\n",
    "\n",
    "module load apps/python/conda\n",
    "\n",
    "source activate myspark\n",
    "\n",
    "spark-submit ../Code/LogMiningBig.py\n",
    "~~~~\n",
    "\n",
    "* Get necessary files on your ShARC.\n",
    "* Start a session with command `qrshx`\n",
    "* Under appopriate directory submit yur job via the `qsub` comand\n",
    "\n",
    "`qsub Lab1_SubmitBatch.sh`\n",
    "\n",
    "Check the status of your quening/running job(s) `qstat` (jobs not shown are finished already).\n",
    "\n",
    "Check your output file, which is **`COM6012_Lab1.output`** specified with option **`-o`** above. You can change it to a name you like."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Exercise\n",
    "\n",
    "### More mining questions (completing three or more questions is considered as completion of this exercise):\n",
    "\n",
    "#### Easier questions (recommended)\n",
    "* How many requests in total?\n",
    "* How many requests on a particular day (e.g., 15th August)?\n",
    "* How many 404 (page not found) errors in total?\n",
    "* How many 404 (page not found) errors on a particular day (e.g., 15th August)?\n",
    "* How many requests from a particular host (e.g.,uplherc.up.com)?\n",
    "* Any other question that you are interested in.\n",
    "\n",
    "#### More challenging questions that will become easier to answer in Session 2 (optional for Session 1)\n",
    "* How many **unique** hosts on a particular day (e.g., 15th August)?\n",
    "* How many **unique** hosts in total (i.e., in August 1995)?\n",
    "* Which host is the most frequent visitor?\n",
    "* How many different types of return codes?\n",
    "* How many requests per day on average?\n",
    "* How many requests per post on average?\n",
    "* Any other question that you are interested in.\n",
    "\n",
    "### The effects of caching (recommended)\n",
    "* **Compare** the time taken to complete your jobs **with and without** `cache()`.\n",
    "\n",
    "# Acknowledgements\n",
    "\n",
    "Many thanks to Twin, Will, and Mike for their kind help and all those kind contributors of open resources.\n",
    "\n",
    "The log mining problem is adapted from [UC Berkeley cs105x L3](https://www.edx.org/course/introduction-apache-spark-uc-berkeleyx-cs105x)."
   ]
  }
 ],
 "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}