Assignment 3.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h1>Intialization For Spark</h1>\n",
    "\n",
    "We use *findspark* and *pyspark* packages for the initialization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "import pyspark # only run after findspark.init()\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 1</h3>\n",
    "\n",
    "In these commands, we load the data from the './data' folder.\n",
    "\n",
    "We load the 'retail-data' data, which are CSV files and specifying the format by the use of 'read.format' method. Then we specify options which are, *header=true* in order to use the first row in the CSV files as the dataframe columns names (headers), *inferSchema=True* which make spark automatically detects the columns data types. After that we specify the location of the CSVs with the function *load*, we use *coalesce(5)* to reduce the number of nodes to 5 and to avoid full shuffling, then we use *where* to filter the data to get the ones which does not has a NULL value in the *Description* field.\n",
    "\n",
    "Then we load the the 'simple-ml' json data using 'spark.read.json' and specifying the location.\n",
    "\n",
    "After that we load the 'simple-ml-scaling' parquet data using 'spark.read.parquet' and specifying the location."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# COMMANDS 1 :-\n",
    "#------------\n",
    "sales = spark.read.format(\"csv\")\\\n",
    ".option(\"header\", \"true\")\\\n",
    ".option(\"inferSchema\", \"true\")\\\n",
    ".load(\"./data/retail-data/by-day/*.csv\")\\\n",
    ".coalesce(5)\\\n",
    ".where(\"Description IS NOT NULL\")\n",
    "simpleDF = spark.read.json(\"./data/simple-ml\")\n",
    "scaleDF = spark.read.parquet(\"./data/simple-ml-scaling\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 2</h3>\n",
    "\n",
    "In these commands, we use *StandardScaler* to scale *scaleDF* features vector to a new vector which values are in similar scale(removing the mean and scaling to unit std). Then we display the transformation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------+-----------------------------------+\n",
      "| id|      features|StandardScaler_0ef1b26ebad7__output|\n",
      "+---+--------------+-----------------------------------+\n",
      "|  0|[1.0,0.1,-1.0]|               [1.19522860933439...|\n",
      "|  1| [2.0,1.1,1.0]|               [2.39045721866878...|\n",
      "|  0|[1.0,0.1,-1.0]|               [1.19522860933439...|\n",
      "|  1| [2.0,1.1,1.0]|               [2.39045721866878...|\n",
      "|  1|[3.0,10.1,3.0]|               [3.58568582800318...|\n",
      "+---+--------------+-----------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# COMMANDS 2 :-\n",
    "#------------\n",
    "from pyspark.ml.feature import StandardScaler\n",
    "sScaler = StandardScaler().setInputCol(\"features\")\n",
    "sScaler.fit(scaleDF).transform(scaleDF).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 3</h3>\n",
    "\n",
    "In these commands, we use *MinMaxScaler* to scale *scaleDF* features vector to a new vector which values are in a scale of a minimum value of 5 and a maximum value of 10. Then we display the transformation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------+---------------------------------+\n",
      "| id|      features|MinMaxScaler_2b7b9cd20567__output|\n",
      "+---+--------------+---------------------------------+\n",
      "|  0|[1.0,0.1,-1.0]|                    [5.0,5.0,5.0]|\n",
      "|  1| [2.0,1.1,1.0]|                    [7.5,5.5,7.5]|\n",
      "|  0|[1.0,0.1,-1.0]|                    [5.0,5.0,5.0]|\n",
      "|  1| [2.0,1.1,1.0]|                    [7.5,5.5,7.5]|\n",
      "|  1|[3.0,10.1,3.0]|                 [10.0,10.0,10.0]|\n",
      "+---+--------------+---------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# COMMANDS 3 :-\n",
    "#------------\n",
    "from pyspark.ml.feature import MinMaxScaler\n",
    "minMax = MinMaxScaler().setMin(5).setMax(10).setInputCol(\"features\")\n",
    "fittedminMax = minMax.fit(scaleDF)\n",
    "fittedminMax.transform(scaleDF).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 4</h3>\n",
    "\n",
    "In these commands, we use *Normalizer* to normalize *scaleDF* vectors to have unit norm, with a p-norm of 1. Then we display the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------+-------------------------------+\n",
      "| id|      features|Normalizer_c5349cfdcceb__output|\n",
      "+---+--------------+-------------------------------+\n",
      "|  0|[1.0,0.1,-1.0]|           [0.47619047619047...|\n",
      "|  1| [2.0,1.1,1.0]|           [0.48780487804878...|\n",
      "|  0|[1.0,0.1,-1.0]|           [0.47619047619047...|\n",
      "|  1| [2.0,1.1,1.0]|           [0.48780487804878...|\n",
      "|  1|[3.0,10.1,3.0]|           [0.18633540372670...|\n",
      "+---+--------------+-------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# COMMANDS 4 :-\n",
    "#------------\n",
    "from pyspark.ml.feature import Normalizer\n",
    "manhattanDistance = Normalizer().setP(1).setInputCol(\"features\")\n",
    "manhattanDistance.transform(scaleDF).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 5</h3>\n",
    "\n",
    "In these commands, we use *StringIndexer* to encode the *lab* feature in *simpleDF* (which is of type string) into numbers representing each different string value in the coulmn *labelInd* (using default ordering, which is “frequencyDesc”). Then we display the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+----+------+------------------+--------+\n",
      "|color| lab|value1|            value2|labelInd|\n",
      "+-----+----+------+------------------+--------+\n",
      "|green|good|     1|14.386294994851129|     1.0|\n",
      "| blue| bad|     8|14.386294994851129|     0.0|\n",
      "| blue| bad|    12|14.386294994851129|     0.0|\n",
      "|green|good|    15| 38.97187133755819|     1.0|\n",
      "|green|good|    12|14.386294994851129|     1.0|\n",
      "|green| bad|    16|14.386294994851129|     0.0|\n",
      "|  red|good|    35|14.386294994851129|     1.0|\n",
      "|  red| bad|     1| 38.97187133755819|     0.0|\n",
      "|  red| bad|     2|14.386294994851129|     0.0|\n",
      "|  red| bad|    16|14.386294994851129|     0.0|\n",
      "|  red|good|    45| 38.97187133755819|     1.0|\n",
      "|green|good|     1|14.386294994851129|     1.0|\n",
      "| blue| bad|     8|14.386294994851129|     0.0|\n",
      "| blue| bad|    12|14.386294994851129|     0.0|\n",
      "|green|good|    15| 38.97187133755819|     1.0|\n",
      "|green|good|    12|14.386294994851129|     1.0|\n",
      "|green| bad|    16|14.386294994851129|     0.0|\n",
      "|  red|good|    35|14.386294994851129|     1.0|\n",
      "|  red| bad|     1| 38.97187133755819|     0.0|\n",
      "|  red| bad|     2|14.386294994851129|     0.0|\n",
      "+-----+----+------+------------------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# COMMANDS 5 :-\n",
    "#------------\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "lblIndxr = StringIndexer().setInputCol(\"lab\").setOutputCol(\"labelInd\")\n",
    "idxRes = lblIndxr.fit(simpleDF).transform(simpleDF)\n",
    "idxRes.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 6</h3>\n",
    "\n",
    "In these commands, we use *StringIndexer* to encode the *value1* feature in *simpleDF* (which is of type int) into numbers representing each different string value (The string cast of each int value) in the coulmn *valueInd* (using default ordering, which is “frequencyDesc”). Then we display the data. What happens is that it counts the frequency of each value in the *value1* column, then order them descendingly, then assign the value for each frequency a number based on its order."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+----+------+------------------+--------+\n",
      "|color| lab|value1|            value2|valueInd|\n",
      "+-----+----+------+------------------+--------+\n",
      "|green|good|     1|14.386294994851129|     2.0|\n",
      "| blue| bad|     8|14.386294994851129|     4.0|\n",
      "| blue| bad|    12|14.386294994851129|     0.0|\n",
      "|green|good|    15| 38.97187133755819|     5.0|\n",
      "|green|good|    12|14.386294994851129|     0.0|\n",
      "|green| bad|    16|14.386294994851129|     1.0|\n",
      "|  red|good|    35|14.386294994851129|     6.0|\n",
      "|  red| bad|     1| 38.97187133755819|     2.0|\n",
      "|  red| bad|     2|14.386294994851129|     7.0|\n",
      "|  red| bad|    16|14.386294994851129|     1.0|\n",
      "|  red|good|    45| 38.97187133755819|     3.0|\n",
      "|green|good|     1|14.386294994851129|     2.0|\n",
      "| blue| bad|     8|14.386294994851129|     4.0|\n",
      "| blue| bad|    12|14.386294994851129|     0.0|\n",
      "|green|good|    15| 38.97187133755819|     5.0|\n",
      "|green|good|    12|14.386294994851129|     0.0|\n",
      "|green| bad|    16|14.386294994851129|     1.0|\n",
      "|  red|good|    35|14.386294994851129|     6.0|\n",
      "|  red| bad|     1| 38.97187133755819|     2.0|\n",
      "|  red| bad|     2|14.386294994851129|     7.0|\n",
      "+-----+----+------+------------------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# COMMANDS 6 :-\n",
    "#------------\n",
    "valIndexer = StringIndexer().setInputCol(\"value1\").setOutputCol(\"valueInd\")\n",
    "valIndexer.fit(simpleDF).transform(simpleDF).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 7</h3>\n",
    "\n",
    "In these commands, we use *StringIndexer* to encode the *color* feature in *simpleDF* (which is of type int) into numbers representing each different string value in the coulmn *colorInd* (using default ordering, which is “frequencyDesc”). After that we use *OneHotEncoder* to map *colorInd* result column to a column of binary vectors,  that indicates the input category index. Then we display the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------+----------------------------------+\n",
      "|color|colorInd|OneHotEncoder_487247a264db__output|\n",
      "+-----+--------+----------------------------------+\n",
      "|green|     1.0|                     (2,[1],[1.0])|\n",
      "| blue|     2.0|                         (2,[],[])|\n",
      "| blue|     2.0|                         (2,[],[])|\n",
      "|green|     1.0|                     (2,[1],[1.0])|\n",
      "|green|     1.0|                     (2,[1],[1.0])|\n",
      "|green|     1.0|                     (2,[1],[1.0])|\n",
      "|  red|     0.0|                     (2,[0],[1.0])|\n",
      "|  red|     0.0|                     (2,[0],[1.0])|\n",
      "|  red|     0.0|                     (2,[0],[1.0])|\n",
      "|  red|     0.0|                     (2,[0],[1.0])|\n",
      "|  red|     0.0|                     (2,[0],[1.0])|\n",
      "|green|     1.0|                     (2,[1],[1.0])|\n",
      "| blue|     2.0|                         (2,[],[])|\n",
      "| blue|     2.0|                         (2,[],[])|\n",
      "|green|     1.0|                     (2,[1],[1.0])|\n",
      "|green|     1.0|                     (2,[1],[1.0])|\n",
      "|green|     1.0|                     (2,[1],[1.0])|\n",
      "|  red|     0.0|                     (2,[0],[1.0])|\n",
      "|  red|     0.0|                     (2,[0],[1.0])|\n",
      "|  red|     0.0|                     (2,[0],[1.0])|\n",
      "+-----+--------+----------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# COMMANDS 7 :-\n",
    "#------------\n",
    "from pyspark.ml.feature import OneHotEncoder, StringIndexer\n",
    "lblIndxr = StringIndexer().setInputCol(\"color\").setOutputCol(\"colorInd\")\n",
    "colorLab = lblIndxr.fit(simpleDF).transform(simpleDF.select(\"color\"))\n",
    "ohe = OneHotEncoder().setInputCol(\"colorInd\")\n",
    "ohe.transform(colorLab).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 8</h3>\n",
    "\n",
    "In these commands, we use *RegexTokenizer* to extract tokens (tokenize) from the feature *Description* in *sales* dataframe following the pattern of (\" \") which is everything seperated by a space, then make it lowercase and save the tokens into the coulmn *DescOut*. Then we display the first 20 columns of the result and the values are not truncated (*truncate=False*)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------+------------------------------------------+\n",
      "|Description                        |DescOut                                   |\n",
      "+-----------------------------------+------------------------------------------+\n",
      "|RABBIT NIGHT LIGHT                 |[rabbit, night, light]                    |\n",
      "|DOUGHNUT LIP GLOSS                 |[doughnut, lip, gloss]                    |\n",
      "|12 MESSAGE CARDS WITH ENVELOPES    |[12, message, cards, with, envelopes]     |\n",
      "|BLUE HARMONICA IN BOX              |[blue, harmonica, in, box]                |\n",
      "|GUMBALL COAT RACK                  |[gumball, coat, rack]                     |\n",
      "|SKULLS  WATER TRANSFER TATTOOS     |[skulls, water, transfer, tattoos]        |\n",
      "|FELTCRAFT GIRL AMELIE KIT          |[feltcraft, girl, amelie, kit]            |\n",
      "|CAMOUFLAGE LED TORCH               |[camouflage, led, torch]                  |\n",
      "|WHITE SKULL HOT WATER BOTTLE       |[white, skull, hot, water, bottle]        |\n",
      "|ENGLISH ROSE HOT WATER BOTTLE      |[english, rose, hot, water, bottle]       |\n",
      "|HOT WATER BOTTLE KEEP CALM         |[hot, water, bottle, keep, calm]          |\n",
      "|SCOTTIE DOG HOT WATER BOTTLE       |[scottie, dog, hot, water, bottle]        |\n",
      "|ROSE CARAVAN DOORSTOP              |[rose, caravan, doorstop]                 |\n",
      "|GINGHAM HEART  DOORSTOP RED        |[gingham, heart, doorstop, red]           |\n",
      "|STORAGE TIN VINTAGE LEAF           |[storage, tin, vintage, leaf]             |\n",
      "|SET OF 4 KNICK KNACK TINS POPPIES  |[set, of, 4, knick, knack, tins, poppies] |\n",
      "|POPCORN HOLDER                     |[popcorn, holder]                         |\n",
      "|GROW A FLYTRAP OR SUNFLOWER IN TIN |[grow, a, flytrap, or, sunflower, in, tin]|\n",
      "|AIRLINE BAG VINTAGE WORLD CHAMPION |[airline, bag, vintage, world, champion]  |\n",
      "|AIRLINE BAG VINTAGE JET SET BROWN  |[airline, bag, vintage, jet, set, brown]  |\n",
      "+-----------------------------------+------------------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# COMMANDS 8 :-\n",
    "#------------\n",
    "from pyspark.ml.feature import RegexTokenizer\n",
    "rt = RegexTokenizer()\\\n",
    ".setInputCol(\"Description\")\\\n",
    ".setOutputCol(\"DescOut\")\\\n",
    ".setPattern(\" \")\\\n",
    ".setToLowercase(True)\n",
    "rt.transform(sales.select(\"Description\")).show(20, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Commands 9</h3>\n",
    "\n",
    "We can not execute these commands, as the *fittedCV* variable is not defined."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'fittedCV' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-42-045548969fc8>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      6\u001b[0m \u001b[0;34m.\u001b[0m\u001b[0mtransform\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msales\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mselect\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Description\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"CustomerId\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      7\u001b[0m \u001b[0;34m.\u001b[0m\u001b[0mwhere\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"CustomerId IS NOT NULL\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 8\u001b[0;31m \u001b[0mprechi\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mfittedCV\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtransform\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtokenized\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      9\u001b[0m \u001b[0;34m.\u001b[0m\u001b[0mwhere\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"CustomerId IS NOT NULL\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     10\u001b[0m \u001b[0mchisq\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mChiSqSelector\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mNameError\u001b[0m: name 'fittedCV' is not defined"
     ]
    }
   ],
   "source": [
    "# COMMANDS 9 :-\n",
    "#------------\n",
    "from pyspark.ml.feature import ChiSqSelector, Tokenizer\n",
    "tkn = Tokenizer().setInputCol(\"Description\").setOutputCol(\"DescOut\")\n",
    "tokenized = tkn\\\n",
    ".transform(sales.select(\"Description\", \"CustomerId\"))\\\n",
    ".where(\"CustomerId IS NOT NULL\")\n",
    "prechi = fittedCV.transform(tokenized)\\\n",
    ".where(\"CustomerId IS NOT NULL\")\n",
    "chisq = ChiSqSelector()\\\n",
    ".setFeaturesCol(\"countVec\")\\\n",
    ".setLabelCol(\"CustomerId\")\\\n",
    ".setNumTopFeatures(2)\n",
    "chisq.fit(prechi).transform(prechi)\\\n",
    ".drop(\"customerId\", \"Description\", \"DescOut\").show()"
   ]
  },
  {
   "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}