PySpark-Retaildata.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark in the context of sales data "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sales and customer data provide important insight for companies. Analytics has today an important role in providing data about sales trends, customer profiles, warnings about (un)profitable products, sales volume forecasts and stock supply as well as personalizing customer relationships and giving product recommendations. All this information needs to be delivered in time, preferably in a (semi)automated fashion.\n",
    "\n",
    "These data can become large to the point that analytical tools and software have performance issues. For example, many businesses embrace spreadsheet software like Excel for understandable reasons, but it would breakdown with large data. Many data scientist work with Python/R, but modules like Pandas would become slow and run out of memory with large data as well.\n",
    "\n",
    "Apache Spark enables large and big data analyses. It does this by using parallel processing using different threads and cores optimally. It can therefore improve performance on a cluster but also on a single machine [1]. It can do this for (i) unstructured data such as text or for (ii) structured data such as DataFrames arranged in columns. Spark does not require loading all data into memory before processing and it is faster than for example Hadoop. \n",
    "Spark is a multi-language tool. Interfaces exist for Scala, Java, Python and R users, and it can be used in the cloud. These and other features make it a suitable platform for large scale data analyses. Google trends suggests that PySpark -- Spark with a Python interface -- enjoys increasing popularity."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following blog shows a detailed short example using PySpark in the context of the Online retail sales data [2]. These are data that are arranged in column format, containing for example invoice number, invoice dates, quantity, price, product description. The chosen data serve as an example and the size would still work in Pandas for most single machine users, even though it would be slower.\n",
    "\n",
    "The current blog does not provide a benchmark as done previously [1]. It rather gives hands-on analytical steps with code (i.e., concatenate data, removal of data records, renaming columns, replacing strings, casting data types, creation of new features, filtering data). It therefore allows a first glimpse into the world of PySpark. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**IMPORT MODULES**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Jupyter notebook file *PySpark-retaildata.ipynb* can be found at GitHub: https://github.com/RRighart/Retail\n",
    "The notebook can be run in Google Colab. To install PySpark, type in the first cell `!pip install pyspark`. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "An alternative is using Docker. A PySpark notebook can be started with https://github.com/jupyter/docker-stacks/tree/master/pyspark-notebook .\n",
    "\n",
    "After PySpark is installed and the Jupyter notebook is up and running, we first need import the modules and create a Spark session:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import json\n",
    "import pyspark.sql.functions as F\n",
    "import pyspark.sql\n",
    "from pyspark.sql.functions import col, skewness, kurtosis\n",
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql.functions import * \n",
    "from pyspark.sql.functions import isnan, when, count, col\n",
    "from pyspark.sql.functions import when\n",
    "from pyspark.sql.functions import UserDefinedFunction\n",
    "from pyspark.sql.functions import from_unixtime, unix_timestamp\n",
    "from pyspark.sql.types import StringType\n",
    "from datetime import date, timedelta, datetime\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'2.4.5'"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that the Spark version used for here is 2.4.5, which can be found by the command `spark.version`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**LOADING AND DISPLAYING THE DATA**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Online retail data [2] can be downloaded from http://archive.ics.uci.edu/ml/machine-learning-databases/00502/ . The data sheets should be converted to online1.csv and online2.csv to facilitate loading from disk. The command `pwd` or `os.getcwd()` can be used to find the current directory from which PySpark will load the files. Below it can be seen that PySpark only takes a couple of seconds whereas Pandas would take a couple of minutes on the same machine. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_data(data):\n",
    "    \"\"\"data=\"online1.csv\" or \"online2.csv\" \"\"\"\n",
    "    t1=time.time()\n",
    "    dat = spark.read.options(header=True, inferSchema=True).csv(data)\n",
    "    t2=time.time()\n",
    "    print(\"Duration:\", np.round((t2-t1), 2), \"seconds\")\n",
    "    return(dat)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Duration: 5.44 seconds\n"
     ]
    }
   ],
   "source": [
    "df1 = load_data('online1.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Duration: 1.76 seconds\n"
     ]
    }
   ],
   "source": [
    "df2 = load_data('online2.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Size of the DataFrames (or shapes in terms of Pandas) can be obtained with the following code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data shape (rows, columns): 525461 x 8\n",
      "Data shape (rows, columns): 541910 x 8\n"
     ]
    }
   ],
   "source": [
    "def datashape(data):\n",
    "    print(\"Data shape (rows, columns):\", data.count(), \"x\", len(data.columns))\n",
    "    \n",
    "datashape(df1)\n",
    "datashape(df2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To have all the data together in one DataFrame, df1 and df2 will be concatenated vertically."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data shape (rows, columns): 1067371 x 8\n"
     ]
    }
   ],
   "source": [
    "df = df1.unionByName(df2)\n",
    "\n",
    "datashape(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following displays the first 5 rows. The command `.limit(5)` will be used frequently throughout the text, which is comparable to the equivalent `.head(5)` in Pandas, to set the number of rows that is displayed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+---------+--------------------+--------+--------------+-----+-----------+--------------+\n",
      "|Invoice|StockCode|         Description|Quantity|   InvoiceDate|Price|Customer ID|       Country|\n",
      "+-------+---------+--------------------+--------+--------------+-----+-----------+--------------+\n",
      "| 489434|    85048|15CM CHRISTMAS GL...|      12|12/1/2009 7:45| 6.95|      13085|United Kingdom|\n",
      "| 489434|   79323P|  PINK CHERRY LIGHTS|      12|12/1/2009 7:45| 6.75|      13085|United Kingdom|\n",
      "| 489434|   79323W| WHITE CHERRY LIGHTS|      12|12/1/2009 7:45| 6.75|      13085|United Kingdom|\n",
      "| 489434|    22041|\"RECORD FRAME 7\"\"...|      48|12/1/2009 7:45|  2.1|      13085|United Kingdom|\n",
      "| 489434|    21232|STRAWBERRY CERAMI...|      24|12/1/2009 7:45| 1.25|      13085|United Kingdom|\n",
      "+-------+---------+--------------------+--------+--------------+-----+-----------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.limit(5).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is the traditional Spark DataFrame output.\n",
    "By using the following setting we will get from now on a Pandas-like output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.set(\"spark.sql.repl.eagerEval.enabled\", True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A number of descriptive statistics can be obtained, like count, standard deviation, mean, minimum and maximum."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th>summary</th>\n",
       "      <th>count</th>\n",
       "      <th>mean</th>\n",
       "      <th>stddev</th>\n",
       "      <th>min</th>\n",
       "      <th>max</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>Invoice</th>\n",
       "      <td>1067371</td>\n",
       "      <td>537608.1499316233</td>\n",
       "      <td>26662.4504469045</td>\n",
       "      <td>489434</td>\n",
       "      <td>C581569</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>StockCode</th>\n",
       "      <td>1067371</td>\n",
       "      <td>29011.161534536903</td>\n",
       "      <td>18822.94286618918</td>\n",
       "      <td>10002</td>\n",
       "      <td>m</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>Description</th>\n",
       "      <td>1062989</td>\n",
       "      <td>21848.25</td>\n",
       "      <td>922.9197780233488</td>\n",
       "      <td>DOORMAT UNION JACK GUNS AND ROSES</td>\n",
       "      <td>wrongly sold sets</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>Quantity</th>\n",
       "      <td>1067371</td>\n",
       "      <td>9.9388984711033</td>\n",
       "      <td>172.7057940767536</td>\n",
       "      <td>-80995</td>\n",
       "      <td>80995</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>InvoiceDate</th>\n",
       "      <td>1067371</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1/10/2010 10:26</td>\n",
       "      <td>9/9/2011 9:52</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>Price</th>\n",
       "      <td>1067371</td>\n",
       "      <td>4.649387727417394</td>\n",
       "      <td>123.55305872146346</td>\n",
       "      <td>-53594.36</td>\n",
       "      <td>38970.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>Customer ID</th>\n",
       "      <td>824364</td>\n",
       "      <td>15324.63850435002</td>\n",
       "      <td>1697.4644503793093</td>\n",
       "      <td>12346</td>\n",
       "      <td>18287</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>Country</th>\n",
       "      <td>1067371</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Australia</td>\n",
       "      <td>West Indies</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "summary        count                mean              stddev  \\\n",
       "Invoice      1067371   537608.1499316233    26662.4504469045   \n",
       "StockCode    1067371  29011.161534536903   18822.94286618918   \n",
       "Description  1062989            21848.25   922.9197780233488   \n",
       "Quantity     1067371     9.9388984711033   172.7057940767536   \n",
       "InvoiceDate  1067371                None                None   \n",
       "Price        1067371   4.649387727417394  123.55305872146346   \n",
       "Customer ID   824364   15324.63850435002  1697.4644503793093   \n",
       "Country      1067371                None                None   \n",
       "\n",
       "summary                                      min                max  \n",
       "Invoice                                   489434            C581569  \n",
       "StockCode                                  10002                  m  \n",
       "Description    DOORMAT UNION JACK GUNS AND ROSES  wrongly sold sets  \n",
       "Quantity                                  -80995              80995  \n",
       "InvoiceDate                      1/10/2010 10:26      9/9/2011 9:52  \n",
       "Price                                  -53594.36            38970.0  \n",
       "Customer ID                                12346              18287  \n",
       "Country                                Australia        West Indies  "
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "summary = df.describe().toPandas()\n",
    "summary = summary.T\n",
    "summary.columns = summary.iloc[0]\n",
    "summary = summary.drop(summary.index[0])\n",
    "summary"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The table column *count* suggests that there are missing values in *Description* and *Customer ID*. The number of NaN is displayed by the following code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Invoice</th><th>StockCode</th><th>Description</th><th>Quantity</th><th>InvoiceDate</th><th>Price</th><th>Customer ID</th><th>Country</th></tr>\n",
       "<tr><td>0</td><td>0</td><td>4382</td><td>0</td><td>0</td><td>0</td><td>243007</td><td>0</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------+---------+-----------+--------+-----------+-----+-----------+-------+\n",
       "|Invoice|StockCode|Description|Quantity|InvoiceDate|Price|Customer ID|Country|\n",
       "+-------+---------+-----------+--------+-----------+-----+-----------+-------+\n",
       "|      0|        0|       4382|       0|          0|    0|     243007|      0|\n",
       "+-------+---------+-----------+--------+-----------+-----+-----------+-------+"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df.columns])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cancelled transactions start with a capital C in the column Invoice. These will be removed by the following."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df[~df['Invoice'].startswith(\"C\")]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The datashape function confirms that cancelled transactions have been removed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data shape (rows, columns): 1047877 x 8\n"
     ]
    }
   ],
   "source": [
    "datashape(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**CHANGING NAME AND DATA TYPE**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The columnname *Customer ID* contains an annoying white space that under certain circumstances can cause problems. So we should better rename that."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumnRenamed('Customer ID', 'CustomerID')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Further, the *Country* EIRE can be replaced by Ireland. The result can be verified by `df.filter(df.Country == \"Ireland\").limit(5)`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.replace(['EIRE'],['Ireland'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Information about the column type can be obtained. If you want some supplementary information, an alternative command is `df.explain(df)` ."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType(List(StructField(Invoice,StringType,true),StructField(StockCode,StringType,true),StructField(Description,StringType,true),StructField(Quantity,IntegerType,true),StructField(InvoiceDate,StringType,true),StructField(Price,DoubleType,true),StructField(CustomerID,IntegerType,true),StructField(Country,StringType,true)))"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Knowing the column types, we could now cast the column Quantity from integer to float."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"Quantity\", col(\"Quantity\").cast(\"Float\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The *InvoiceDate* shows up as a string. In order to exploit the time-related information, it would be best to convert it to date type. We will do this by creating a new column *n_InvoiceDate*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"n_InvoiceDate\", from_unixtime(unix_timestamp('InvoiceDate', 'MM/dd/yyyy HH:mm')).alias('n_InvoiceDate'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sorting the new dates, we can now display the number of records (rows) as a function of date and time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>n_InvoiceDate</th><th>count</th></tr>\n",
       "<tr><td>2009-12-01 07:45:00</td><td>8</td></tr>\n",
       "<tr><td>2009-12-01 07:46:00</td><td>4</td></tr>\n",
       "<tr><td>2009-12-01 09:06:00</td><td>19</td></tr>\n",
       "<tr><td>2009-12-01 09:08:00</td><td>23</td></tr>\n",
       "<tr><td>2009-12-01 09:24:00</td><td>17</td></tr>\n",
       "<tr><td>2009-12-01 09:28:00</td><td>19</td></tr>\n",
       "<tr><td>2009-12-01 09:43:00</td><td>2</td></tr>\n",
       "<tr><td>2009-12-01 09:44:00</td><td>4</td></tr>\n",
       "<tr><td>2009-12-01 09:46:00</td><td>23</td></tr>\n",
       "<tr><td>2009-12-01 09:50:00</td><td>7</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------------------+-----+\n",
       "|      n_InvoiceDate|count|\n",
       "+-------------------+-----+\n",
       "|2009-12-01 07:45:00|    8|\n",
       "|2009-12-01 07:46:00|    4|\n",
       "|2009-12-01 09:06:00|   19|\n",
       "|2009-12-01 09:08:00|   23|\n",
       "|2009-12-01 09:24:00|   17|\n",
       "|2009-12-01 09:28:00|   19|\n",
       "|2009-12-01 09:43:00|    2|\n",
       "|2009-12-01 09:44:00|    4|\n",
       "|2009-12-01 09:46:00|   23|\n",
       "|2009-12-01 09:50:00|    7|\n",
       "+-------------------+-----+"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupby(\"n_InvoiceDate\").count().sort(\"n_InvoiceDate\", ascending=True).limit(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It is also possible to display a range of dates. The following spots all sales at 15 Jan 2010 from 8 till 10 o'clock."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Invoice</th><th>StockCode</th><th>Description</th><th>Quantity</th><th>InvoiceDate</th><th>Price</th><th>CustomerID</th><th>Country</th><th>n_InvoiceDate</th></tr>\n",
       "<tr><td>494508</td><td>84380</td><td>SET OF 3 BUTTERFL...</td><td>24.0</td><td>1/15/2010 8:13</td><td>1.25</td><td>16353</td><td>United Kingdom</td><td>2010-01-15 08:13:00</td></tr>\n",
       "<tr><td>494508</td><td>21174</td><td>POTTERING IN THE ...</td><td>12.0</td><td>1/15/2010 8:13</td><td>1.95</td><td>16353</td><td>United Kingdom</td><td>2010-01-15 08:13:00</td></tr>\n",
       "<tr><td>494508</td><td>21041</td><td>RED SPOTTY OVEN G...</td><td>12.0</td><td>1/15/2010 8:13</td><td>2.95</td><td>16353</td><td>United Kingdom</td><td>2010-01-15 08:13:00</td></tr>\n",
       "<tr><td>494508</td><td>21154</td><td>RED SPOTTY OVEN G...</td><td>20.0</td><td>1/15/2010 8:13</td><td>1.25</td><td>16353</td><td>United Kingdom</td><td>2010-01-15 08:13:00</td></tr>\n",
       "<tr><td>494509</td><td>21700</td><td>BIG DOUGHNUT FRID...</td><td>36.0</td><td>1/15/2010 8:14</td><td>0.85</td><td>16353</td><td>United Kingdom</td><td>2010-01-15 08:14:00</td></tr>\n",
       "<tr><td>494509</td><td>84378</td><td>SET OF 3 HEART CO...</td><td>12.0</td><td>1/15/2010 8:14</td><td>1.25</td><td>16353</td><td>United Kingdom</td><td>2010-01-15 08:14:00</td></tr>\n",
       "<tr><td>494510</td><td>21733</td><td>RED HANGING HEART...</td><td>6.0</td><td>1/15/2010 8:59</td><td>2.55</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>85123A</td><td>WHITE HANGING HEA...</td><td>6.0</td><td>1/15/2010 8:59</td><td>2.55</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>15056N</td><td>EDWARDIAN PARASOL...</td><td>6.0</td><td>1/15/2010 8:59</td><td>4.95</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>82483</td><td>WOOD 2 DRAWER CAB...</td><td>4.0</td><td>1/15/2010 8:59</td><td>4.95</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>82486</td><td>WOOD S/3 CABINET ...</td><td>4.0</td><td>1/15/2010 8:59</td><td>6.95</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>82482</td><td>WOODEN PICTURE FR...</td><td>6.0</td><td>1/15/2010 8:59</td><td>2.1</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>82494L</td><td>WOODEN FRAME ANTI...</td><td>6.0</td><td>1/15/2010 8:59</td><td>2.55</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>21523</td><td>FANCY FONT HOME S...</td><td>4.0</td><td>1/15/2010 8:59</td><td>5.95</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>21955</td><td>UNION JACK GUNS &amp;...</td><td>4.0</td><td>1/15/2010 8:59</td><td>5.95</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>15056BL</td><td>EDWARDIAN PARASOL...</td><td>6.0</td><td>1/15/2010 8:59</td><td>4.95</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>20679</td><td>EDWARDIAN PARASOL...</td><td>6.0</td><td>1/15/2010 8:59</td><td>4.95</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>21935</td><td>SUKI  SHOULDER BAG</td><td>10.0</td><td>1/15/2010 8:59</td><td>1.45</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>21934</td><td>SKULL SHOULDER BAG</td><td>10.0</td><td>1/15/2010 8:59</td><td>1.45</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "<tr><td>494510</td><td>85099C</td><td>JUMBO  BAG BAROQU...</td><td>10.0</td><td>1/15/2010 8:59</td><td>1.65</td><td>17850</td><td>United Kingdom</td><td>2010-01-15 08:59:00</td></tr>\n",
       "</table>\n",
       "only showing top 20 rows\n"
      ],
      "text/plain": [
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+\n",
       "|Invoice|StockCode|         Description|Quantity|   InvoiceDate|Price|CustomerID|       Country|      n_InvoiceDate|\n",
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+\n",
       "| 494508|    84380|SET OF 3 BUTTERFL...|    24.0|1/15/2010 8:13| 1.25|     16353|United Kingdom|2010-01-15 08:13:00|\n",
       "| 494508|    21174|POTTERING IN THE ...|    12.0|1/15/2010 8:13| 1.95|     16353|United Kingdom|2010-01-15 08:13:00|\n",
       "| 494508|    21041|RED SPOTTY OVEN G...|    12.0|1/15/2010 8:13| 2.95|     16353|United Kingdom|2010-01-15 08:13:00|\n",
       "| 494508|    21154|RED SPOTTY OVEN G...|    20.0|1/15/2010 8:13| 1.25|     16353|United Kingdom|2010-01-15 08:13:00|\n",
       "| 494509|    21700|BIG DOUGHNUT FRID...|    36.0|1/15/2010 8:14| 0.85|     16353|United Kingdom|2010-01-15 08:14:00|\n",
       "| 494509|    84378|SET OF 3 HEART CO...|    12.0|1/15/2010 8:14| 1.25|     16353|United Kingdom|2010-01-15 08:14:00|\n",
       "| 494510|    21733|RED HANGING HEART...|     6.0|1/15/2010 8:59| 2.55|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|   85123A|WHITE HANGING HEA...|     6.0|1/15/2010 8:59| 2.55|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|   15056N|EDWARDIAN PARASOL...|     6.0|1/15/2010 8:59| 4.95|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|    82483|WOOD 2 DRAWER CAB...|     4.0|1/15/2010 8:59| 4.95|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|    82486|WOOD S/3 CABINET ...|     4.0|1/15/2010 8:59| 6.95|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|    82482|WOODEN PICTURE FR...|     6.0|1/15/2010 8:59|  2.1|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|   82494L|WOODEN FRAME ANTI...|     6.0|1/15/2010 8:59| 2.55|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|    21523|FANCY FONT HOME S...|     4.0|1/15/2010 8:59| 5.95|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|    21955|UNION JACK GUNS &...|     4.0|1/15/2010 8:59| 5.95|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|  15056BL|EDWARDIAN PARASOL...|     6.0|1/15/2010 8:59| 4.95|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|    20679|EDWARDIAN PARASOL...|     6.0|1/15/2010 8:59| 4.95|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|    21935|  SUKI  SHOULDER BAG|    10.0|1/15/2010 8:59| 1.45|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|    21934|  SKULL SHOULDER BAG|    10.0|1/15/2010 8:59| 1.45|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "| 494510|   85099C|JUMBO  BAG BAROQU...|    10.0|1/15/2010 8:59| 1.65|     17850|United Kingdom|2010-01-15 08:59:00|\n",
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+\n",
       "only showing top 20 rows"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df[(df[\"n_InvoiceDate\"]> '2010-01-15 08:00:00') & (df[\"n_InvoiceDate\"]< '2010-01-15 10:00:00') ]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**CONVERTING DATA**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Time-serie analyses often require different time units, like seconds, minutes, hours, days, weeks, months, years. For example, if we want to display the sales per week, we could use the function `weekofyear` that translates the date to week."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"weekofyear\", weekofyear(\"n_InvoiceDate\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Multiple columns can be used in a computation. The total *Amount* a customer spent can be computed by multiplication of the *Price* of a single product with its *Quantity*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Invoice</th><th>StockCode</th><th>Description</th><th>Quantity</th><th>InvoiceDate</th><th>Price</th><th>CustomerID</th><th>Country</th><th>n_InvoiceDate</th><th>weekofyear</th><th>Amount</th></tr>\n",
       "<tr><td>489434</td><td>85048</td><td>15CM CHRISTMAS GL...</td><td>12.0</td><td>12/1/2009 7:45</td><td>6.95</td><td>13085</td><td>United Kingdom</td><td>2009-12-01 07:45:00</td><td>49</td><td>83.4</td></tr>\n",
       "<tr><td>489434</td><td>79323P</td><td>PINK CHERRY LIGHTS</td><td>12.0</td><td>12/1/2009 7:45</td><td>6.75</td><td>13085</td><td>United Kingdom</td><td>2009-12-01 07:45:00</td><td>49</td><td>81.0</td></tr>\n",
       "<tr><td>489434</td><td>79323W</td><td> WHITE CHERRY LIGHTS</td><td>12.0</td><td>12/1/2009 7:45</td><td>6.75</td><td>13085</td><td>United Kingdom</td><td>2009-12-01 07:45:00</td><td>49</td><td>81.0</td></tr>\n",
       "<tr><td>489434</td><td>22041</td><td>&quot;RECORD FRAME 7&quot;&quot;...</td><td>48.0</td><td>12/1/2009 7:45</td><td>2.1</td><td>13085</td><td>United Kingdom</td><td>2009-12-01 07:45:00</td><td>49</td><td>100.80000000000001</td></tr>\n",
       "<tr><td>489434</td><td>21232</td><td>STRAWBERRY CERAMI...</td><td>24.0</td><td>12/1/2009 7:45</td><td>1.25</td><td>13085</td><td>United Kingdom</td><td>2009-12-01 07:45:00</td><td>49</td><td>30.0</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+----------+------------------+\n",
       "|Invoice|StockCode|         Description|Quantity|   InvoiceDate|Price|CustomerID|       Country|      n_InvoiceDate|weekofyear|            Amount|\n",
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+----------+------------------+\n",
       "| 489434|    85048|15CM CHRISTMAS GL...|    12.0|12/1/2009 7:45| 6.95|     13085|United Kingdom|2009-12-01 07:45:00|        49|              83.4|\n",
       "| 489434|   79323P|  PINK CHERRY LIGHTS|    12.0|12/1/2009 7:45| 6.75|     13085|United Kingdom|2009-12-01 07:45:00|        49|              81.0|\n",
       "| 489434|   79323W| WHITE CHERRY LIGHTS|    12.0|12/1/2009 7:45| 6.75|     13085|United Kingdom|2009-12-01 07:45:00|        49|              81.0|\n",
       "| 489434|    22041|\"RECORD FRAME 7\"\"...|    48.0|12/1/2009 7:45|  2.1|     13085|United Kingdom|2009-12-01 07:45:00|        49|100.80000000000001|\n",
       "| 489434|    21232|STRAWBERRY CERAMI...|    24.0|12/1/2009 7:45| 1.25|     13085|United Kingdom|2009-12-01 07:45:00|        49|              30.0|\n",
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+----------+------------------+"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = df.withColumn(\"Amount\", col(\"Quantity\") * col(\"Price\"))\n",
    "df.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**FILTERING DATA**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "How to select rows that contain specific products? We can use the command `isin`, which is very similar to the Pandas isin function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Invoice</th><th>StockCode</th><th>Description</th><th>Quantity</th><th>InvoiceDate</th><th>Price</th><th>CustomerID</th><th>Country</th><th>n_InvoiceDate</th><th>weekofyear</th><th>Amount</th></tr>\n",
       "<tr><td>489442</td><td>85123A</td><td>WHITE HANGING HEA...</td><td>6.0</td><td>12/1/2009 9:46</td><td>2.95</td><td>13635</td><td>United Kingdom</td><td>2009-12-01 09:46:00</td><td>49</td><td>17.700000000000003</td></tr>\n",
       "<tr><td>489446</td><td>85123A</td><td>WHITE HANGING HEA...</td><td>32.0</td><td>12/1/2009 10:06</td><td>2.55</td><td>13758</td><td>United Kingdom</td><td>2009-12-01 10:06:00</td><td>49</td><td>81.6</td></tr>\n",
       "<tr><td>489465</td><td>85123A</td><td>WHITE HANGING HEA...</td><td>64.0</td><td>12/1/2009 10:52</td><td>2.55</td><td>13767</td><td>United Kingdom</td><td>2009-12-01 10:52:00</td><td>49</td><td>163.2</td></tr>\n",
       "<tr><td>489517</td><td>85123A</td><td>WHITE HANGING HEA...</td><td>1.0</td><td>12/1/2009 11:34</td><td>2.95</td><td>16329</td><td>United Kingdom</td><td>2009-12-01 11:34:00</td><td>49</td><td>2.95</td></tr>\n",
       "<tr><td>489519</td><td>85123A</td><td>WHITE HANGING HEA...</td><td>32.0</td><td>12/1/2009 11:37</td><td>2.55</td><td>17700</td><td>United Kingdom</td><td>2009-12-01 11:37:00</td><td>49</td><td>81.6</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------+---------+--------------------+--------+---------------+-----+----------+--------------+-------------------+----------+------------------+\n",
       "|Invoice|StockCode|         Description|Quantity|    InvoiceDate|Price|CustomerID|       Country|      n_InvoiceDate|weekofyear|            Amount|\n",
       "+-------+---------+--------------------+--------+---------------+-----+----------+--------------+-------------------+----------+------------------+\n",
       "| 489442|   85123A|WHITE HANGING HEA...|     6.0| 12/1/2009 9:46| 2.95|     13635|United Kingdom|2009-12-01 09:46:00|        49|17.700000000000003|\n",
       "| 489446|   85123A|WHITE HANGING HEA...|    32.0|12/1/2009 10:06| 2.55|     13758|United Kingdom|2009-12-01 10:06:00|        49|              81.6|\n",
       "| 489465|   85123A|WHITE HANGING HEA...|    64.0|12/1/2009 10:52| 2.55|     13767|United Kingdom|2009-12-01 10:52:00|        49|             163.2|\n",
       "| 489517|   85123A|WHITE HANGING HEA...|     1.0|12/1/2009 11:34| 2.95|     16329|United Kingdom|2009-12-01 11:34:00|        49|              2.95|\n",
       "| 489519|   85123A|WHITE HANGING HEA...|    32.0|12/1/2009 11:37| 2.55|     17700|United Kingdom|2009-12-01 11:37:00|        49|              81.6|\n",
       "+-------+---------+--------------------+--------+---------------+-----+----------+--------------+-------------------+----------+------------------+"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df[df.Description.isin('WHITE HANGING HEART T-LIGHT HOLDER')].limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we want to search our data by key word, we would use the command `like`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Invoice</th><th>StockCode</th><th>Description</th><th>Quantity</th><th>InvoiceDate</th><th>Price</th><th>CustomerID</th><th>Country</th><th>n_InvoiceDate</th><th>weekofyear</th><th>Amount</th></tr>\n",
       "<tr><td>489434</td><td>79323W</td><td> WHITE CHERRY LIGHTS</td><td>12.0</td><td>12/1/2009 7:45</td><td>6.75</td><td>13085</td><td>United Kingdom</td><td>2009-12-01 07:45:00</td><td>49</td><td>81.0</td></tr>\n",
       "<tr><td>489436</td><td>22142</td><td>CHRISTMAS CRAFT W...</td><td>12.0</td><td>12/1/2009 9:06</td><td>1.45</td><td>13078</td><td>United Kingdom</td><td>2009-12-01 09:06:00</td><td>49</td><td>17.4</td></tr>\n",
       "<tr><td>489436</td><td>21333</td><td>CLASSIC WHITE FRAME</td><td>6.0</td><td>12/1/2009 9:06</td><td>2.95</td><td>13078</td><td>United Kingdom</td><td>2009-12-01 09:06:00</td><td>49</td><td>17.700000000000003</td></tr>\n",
       "<tr><td>489439</td><td>85014B</td><td>RED/WHITE DOTS RU...</td><td>3.0</td><td>12/1/2009 9:28</td><td>5.95</td><td>12682</td><td>France</td><td>2009-12-01 09:28:00</td><td>49</td><td>17.85</td></tr>\n",
       "<tr><td>489441</td><td>84029E</td><td>RED WOOLLY HOTTIE...</td><td>36.0</td><td>12/1/2009 9:44</td><td>2.95</td><td>18087</td><td>United Kingdom</td><td>2009-12-01 09:44:00</td><td>49</td><td>106.2</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+----------+------------------+\n",
       "|Invoice|StockCode|         Description|Quantity|   InvoiceDate|Price|CustomerID|       Country|      n_InvoiceDate|weekofyear|            Amount|\n",
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+----------+------------------+\n",
       "| 489434|   79323W| WHITE CHERRY LIGHTS|    12.0|12/1/2009 7:45| 6.75|     13085|United Kingdom|2009-12-01 07:45:00|        49|              81.0|\n",
       "| 489436|    22142|CHRISTMAS CRAFT W...|    12.0|12/1/2009 9:06| 1.45|     13078|United Kingdom|2009-12-01 09:06:00|        49|              17.4|\n",
       "| 489436|    21333| CLASSIC WHITE FRAME|     6.0|12/1/2009 9:06| 2.95|     13078|United Kingdom|2009-12-01 09:06:00|        49|17.700000000000003|\n",
       "| 489439|   85014B|RED/WHITE DOTS RU...|     3.0|12/1/2009 9:28| 5.95|     12682|        France|2009-12-01 09:28:00|        49|             17.85|\n",
       "| 489441|   84029E|RED WOOLLY HOTTIE...|    36.0|12/1/2009 9:44| 2.95|     18087|United Kingdom|2009-12-01 09:44:00|        49|             106.2|\n",
       "+-------+---------+--------------------+--------+--------------+-----+----------+--------------+-------------------+----------+------------------+"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df[df.Description.like('%WHITE%')].limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To find big buyers, probably organizational customers, select rows where *Quantity* is larger than 50000. Note that `df.where(df.Quantity > 50000)` would give a similar result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Invoice</th><th>StockCode</th><th>Description</th><th>Quantity</th><th>InvoiceDate</th><th>Price</th><th>CustomerID</th><th>Country</th><th>n_InvoiceDate</th><th>weekofyear</th><th>Amount</th></tr>\n",
       "<tr><td>541431</td><td>23166</td><td>MEDIUM CERAMIC TO...</td><td>74215.0</td><td>1/18/2011 10:01</td><td>1.04</td><td>12346</td><td>United Kingdom</td><td>2011-01-18 10:01:00</td><td>3</td><td>77183.6</td></tr>\n",
       "<tr><td>581483</td><td>23843</td><td>PAPER CRAFT , LIT...</td><td>80995.0</td><td>12/9/2011 9:15</td><td>2.08</td><td>16446</td><td>United Kingdom</td><td>2011-12-09 09:15:00</td><td>49</td><td>168469.6</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------+---------+--------------------+--------+---------------+-----+----------+--------------+-------------------+----------+--------+\n",
       "|Invoice|StockCode|         Description|Quantity|    InvoiceDate|Price|CustomerID|       Country|      n_InvoiceDate|weekofyear|  Amount|\n",
       "+-------+---------+--------------------+--------+---------------+-----+----------+--------------+-------------------+----------+--------+\n",
       "| 541431|    23166|MEDIUM CERAMIC TO...| 74215.0|1/18/2011 10:01| 1.04|     12346|United Kingdom|2011-01-18 10:01:00|         3| 77183.6|\n",
       "| 581483|    23843|PAPER CRAFT , LIT...| 80995.0| 12/9/2011 9:15| 2.08|     16446|United Kingdom|2011-12-09 09:15:00|        49|168469.6|\n",
       "+-------+---------+--------------------+--------+---------------+-----+----------+--------------+-------------------+----------+--------+"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.filter(df.Quantity > 50000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let us count the number of data records per country and sort the output, which shows that UK is clearly leading the list."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Country</th><th>count</th></tr>\n",
       "<tr><td>United Kingdom</td><td>964680</td></tr>\n",
       "<tr><td>Ireland</td><td>17354</td></tr>\n",
       "<tr><td>Germany</td><td>16703</td></tr>\n",
       "<tr><td>France</td><td>13941</td></tr>\n",
       "<tr><td>Netherlands</td><td>5093</td></tr>\n",
       "<tr><td>Spain</td><td>3720</td></tr>\n",
       "<tr><td>Switzerland</td><td>3137</td></tr>\n",
       "<tr><td>Belgium</td><td>3069</td></tr>\n",
       "<tr><td>Portugal</td><td>2562</td></tr>\n",
       "<tr><td>Australia</td><td>1815</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------+------+\n",
       "|       Country| count|\n",
       "+--------------+------+\n",
       "|United Kingdom|964680|\n",
       "|       Ireland| 17354|\n",
       "|       Germany| 16703|\n",
       "|        France| 13941|\n",
       "|   Netherlands|  5093|\n",
       "|         Spain|  3720|\n",
       "|   Switzerland|  3137|\n",
       "|       Belgium|  3069|\n",
       "|      Portugal|  2562|\n",
       "|     Australia|  1815|\n",
       "+--------------+------+"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy(\"Country\").count().sort(\"count\", ascending=False).limit(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sort can be used for *InvoiceDate* as well. This can show which hours customers purchase preferentially."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>n_InvoiceDate</th><th>count</th></tr>\n",
       "<tr><td>2009-12-01 07:45:00</td><td>8</td></tr>\n",
       "<tr><td>2009-12-01 07:46:00</td><td>4</td></tr>\n",
       "<tr><td>2009-12-01 09:06:00</td><td>19</td></tr>\n",
       "<tr><td>2009-12-01 09:08:00</td><td>23</td></tr>\n",
       "<tr><td>2009-12-01 09:24:00</td><td>17</td></tr>\n",
       "<tr><td>2009-12-01 09:28:00</td><td>19</td></tr>\n",
       "<tr><td>2009-12-01 09:43:00</td><td>2</td></tr>\n",
       "<tr><td>2009-12-01 09:44:00</td><td>4</td></tr>\n",
       "<tr><td>2009-12-01 09:46:00</td><td>23</td></tr>\n",
       "<tr><td>2009-12-01 09:50:00</td><td>7</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------------------+-----+\n",
       "|      n_InvoiceDate|count|\n",
       "+-------------------+-----+\n",
       "|2009-12-01 07:45:00|    8|\n",
       "|2009-12-01 07:46:00|    4|\n",
       "|2009-12-01 09:06:00|   19|\n",
       "|2009-12-01 09:08:00|   23|\n",
       "|2009-12-01 09:24:00|   17|\n",
       "|2009-12-01 09:28:00|   19|\n",
       "|2009-12-01 09:43:00|    2|\n",
       "|2009-12-01 09:44:00|    4|\n",
       "|2009-12-01 09:46:00|   23|\n",
       "|2009-12-01 09:50:00|    7|\n",
       "+-------------------+-----+"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupby(\"n_InvoiceDate\").count().sort(\"n_InvoiceDate\", ascending=True).limit(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**USING SQL IN PYSPARK**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Luckily, Spark supports SQL - Structured Query Language - which traditionally has an important role in managing relational databases. Using for example SQL queries, a subset of the data can be exported to CRM and KPI sales software. This offers a lot of flexibility for data analyses. Let us experiment with some very useful SQL queries, such as select and filter. We first need to register the DataFrame as a temporary table in the SQLContext. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.registerTempTable(\"df\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We first select a couple of columns, for example *Description* and *Quantity*. If you want to select all columns, simply use the star: `spark.sql(\"select * from df\").show(3)`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Description</th><th>Quantity</th></tr>\n",
       "<tr><td>15CM CHRISTMAS GL...</td><td>12.0</td></tr>\n",
       "<tr><td>PINK CHERRY LIGHTS</td><td>12.0</td></tr>\n",
       "<tr><td> WHITE CHERRY LIGHTS</td><td>12.0</td></tr>\n",
       "<tr><td>&quot;RECORD FRAME 7&quot;&quot;...</td><td>48.0</td></tr>\n",
       "<tr><td>STRAWBERRY CERAMI...</td><td>24.0</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+--------+\n",
       "|         Description|Quantity|\n",
       "+--------------------+--------+\n",
       "|15CM CHRISTMAS GL...|    12.0|\n",
       "|  PINK CHERRY LIGHTS|    12.0|\n",
       "| WHITE CHERRY LIGHTS|    12.0|\n",
       "|\"RECORD FRAME 7\"\"...|    48.0|\n",
       "|STRAWBERRY CERAMI...|    24.0|\n",
       "+--------------------+--------+"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select Description, Quantity from df\").limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Select the columns *Description* and *Quantity* and only those rows where *Quantity* has value = 6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Description</th><th>Quantity</th></tr>\n",
       "<tr><td>CLASSIC WHITE FRAME</td><td>6.0</td></tr>\n",
       "<tr><td>CHRISTMAS CRAFT H...</td><td>6.0</td></tr>\n",
       "<tr><td>CHRISTMAS CRAFT H...</td><td>6.0</td></tr>\n",
       "<tr><td>STRIPES DESIGN MO...</td><td>6.0</td></tr>\n",
       "<tr><td>FELTCRAFT DOLL ROSIE</td><td>6.0</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+--------+\n",
       "|         Description|Quantity|\n",
       "+--------------------+--------+\n",
       "| CLASSIC WHITE FRAME|     6.0|\n",
       "|CHRISTMAS CRAFT H...|     6.0|\n",
       "|CHRISTMAS CRAFT H...|     6.0|\n",
       "|STRIPES DESIGN MO...|     6.0|\n",
       "|FELTCRAFT DOLL ROSIE|     6.0|\n",
       "+--------------------+--------+"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select Description, Quantity from df where Quantity = 6\").limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Select the columns *Description*, *Quantity*, and *Country* where *Quantity* has value = 6 and *country* is United Kingdom."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Description</th><th>Quantity</th><th>Country</th></tr>\n",
       "<tr><td>CLASSIC WHITE FRAME</td><td>6.0</td><td>United Kingdom</td></tr>\n",
       "<tr><td>CHRISTMAS CRAFT H...</td><td>6.0</td><td>United Kingdom</td></tr>\n",
       "<tr><td>CHRISTMAS CRAFT H...</td><td>6.0</td><td>United Kingdom</td></tr>\n",
       "<tr><td>STRIPES DESIGN MO...</td><td>6.0</td><td>United Kingdom</td></tr>\n",
       "<tr><td>FELTCRAFT DOLL ROSIE</td><td>6.0</td><td>United Kingdom</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+--------+--------------+\n",
       "|         Description|Quantity|       Country|\n",
       "+--------------------+--------+--------------+\n",
       "| CLASSIC WHITE FRAME|     6.0|United Kingdom|\n",
       "|CHRISTMAS CRAFT H...|     6.0|United Kingdom|\n",
       "|CHRISTMAS CRAFT H...|     6.0|United Kingdom|\n",
       "|STRIPES DESIGN MO...|     6.0|United Kingdom|\n",
       "|FELTCRAFT DOLL ROSIE|     6.0|United Kingdom|\n",
       "+--------------------+--------+--------------+"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select Description, Quantity, Country from df where Quantity=6 AND Country = 'United Kingdom'\").limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "SQL can also be used to show distinct (unique) values in a column. To limit space, only five are displayed here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Country</th></tr>\n",
       "<tr><td>Sweden</td></tr>\n",
       "<tr><td>Singapore</td></tr>\n",
       "<tr><td>Germany</td></tr>\n",
       "<tr><td>RSA</td></tr>\n",
       "<tr><td>France</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+---------+\n",
       "|  Country|\n",
       "+---------+\n",
       "|   Sweden|\n",
       "|Singapore|\n",
       "|  Germany|\n",
       "|      RSA|\n",
       "|   France|\n",
       "+---------+"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"SELECT DISTINCT Country from df\").limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And we can count the number of distinct values, that is how many countries are in total in the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>count(DISTINCT Country)</th></tr>\n",
       "<tr><td>43</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-----------------------+\n",
       "|count(DISTINCT Country)|\n",
       "+-----------------------+\n",
       "|                     43|\n",
       "+-----------------------+"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"SELECT COUNT(DISTINCT Country) from df\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using SQL we can also exclude certain values. For example, exclude all records with United Kingdom. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Description</th><th>Quantity</th><th>Country</th></tr>\n",
       "<tr><td>SET OF THREE VINT...</td><td>6.0</td><td>France</td></tr>\n",
       "<tr><td>SET/3 RUSSIAN DOL...</td><td>6.0</td><td>France</td></tr>\n",
       "<tr><td>CREAM FELT EASTER...</td><td>6.0</td><td>Australia</td></tr>\n",
       "<tr><td>POTTING SHED TWINE</td><td>6.0</td><td>Australia</td></tr>\n",
       "<tr><td>FOUR HOOK  WHITE ...</td><td>6.0</td><td>Ireland</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+--------+---------+\n",
       "|         Description|Quantity|  Country|\n",
       "+--------------------+--------+---------+\n",
       "|SET OF THREE VINT...|     6.0|   France|\n",
       "|SET/3 RUSSIAN DOL...|     6.0|   France|\n",
       "|CREAM FELT EASTER...|     6.0|Australia|\n",
       "|  POTTING SHED TWINE|     6.0|Australia|\n",
       "|FOUR HOOK  WHITE ...|     6.0|  Ireland|\n",
       "+--------------------+--------+---------+"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select Description, Quantity, Country from df where Quantity=6 AND NOT Country = 'United Kingdom'\").limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It would be possible to add a new column that categorizes UK (1) or not-UK (0). You could then use `df.filter(df.Country == \"United Kingdom\").limit(5)` and `filter(df.Country == \"France\").limit(5)` to check if the column is correctly coding UK versus not-UK."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('Country_UK', F.lit(0))\n",
    "df = df.withColumn(\"Country_UK\", when(df[\"Country\"] == 'United Kingdom', 1).otherwise(df[\"Country_UK\"]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we could count UK versus not-UK using the new column Country_UK."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Country_UK</th><th>count</th></tr>\n",
       "<tr><td>1</td><td>964680</td></tr>\n",
       "<tr><td>0</td><td>83197</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+----------+------+\n",
       "|Country_UK| count|\n",
       "+----------+------+\n",
       "|         1|964680|\n",
       "|         0| 83197|\n",
       "+----------+------+"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy(\"Country_UK\").count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And compute mean *Amount* and summed *Quantity*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Country_UK</th><th>avg(Amount)</th><th>sum(Quantity)</th></tr>\n",
       "<tr><td>1</td><td>18.360805279469655</td><td>9061154.0</td></tr>\n",
       "<tr><td>0</td><td>37.28039221366158</td><td>2038330.0</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+----------+------------------+-------------+\n",
       "|Country_UK|       avg(Amount)|sum(Quantity)|\n",
       "+----------+------------------+-------------+\n",
       "|         1|18.360805279469655|    9061154.0|\n",
       "|         0| 37.28039221366158|    2038330.0|\n",
       "+----------+------------------+-------------+"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy(\"Country_UK\").agg({\"Amount\": \"mean\", \"Quantity\": \"sum\"})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**FINAL THOUGHTS**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That is it for now! In the current blog I have chosen to write out the code line by line. The code can of course be wrapped in several functions with the advantage that it can be used as a pre-processing script for different datasets, where you can set your own parameters. Further, smaller subsets of data can easily be exported and visualized in for ex. Plotly.\n",
    "\n",
    "There are many other exciting features and developments in Spark:\n",
    "1. Koalas is a Pandas API in Apache Spark, with similar capabilities but in a big data environment. This is particularly good news for people who already work in Pandas and need a quick translation to PySpark of their code.\n",
    "2. Pandas UDF is a new feature that allows parallel processing on Pandas DataFrames.\n",
    "3. There are excellent solutions using PySpark in the cloud. For example, AWS has big data platforms such as Elastic Map Reduce (EMR) that support PySpark.\n",
    "4. Spark streaming allows real-time data analysis.\n",
    "5. MLlib allows scalable machine learning in Spark.\n",
    "6. GraphX enables graph computations.\n",
    "\n",
    "If you have any questions or anything you think is missing in this tutorial, please feel free to share. Any suggestions for topics are welcome!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**REFERENCES**\n",
    "\n",
    "[1] Wang, G., Xin, R., and Damji, J. (2018). Benchmarking Apache Spark on a single node machine. https://databricks.com/blog/2018/05/03/benchmarking-apache-spark-on-a-single-node-machine.html\n",
    "\n",
    "[2] Chen, D., Laing Sain, S., Guo, K. (2012) Data mining for the online retail industry. A case\n",
    "study of RFM model-based customer segmentation using data mining. Journal of database marketing & customer strategy management, 19, 197-208."
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}