04. Structured APIs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Spark Image](https://upload.wikimedia.org/wikipedia/commons/thumb/f/f3/Apache_Spark_logo.svg/1200px-Apache_Spark_logo.svg.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Structured APIs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RDDs vs DataFrames and Datasets\n",
    "\n",
    "![](https://databricks.com/wp-content/uploads/2018/05/rdd-1024x595.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resilient Distributed Dataset (RDD)\n",
    "RDD was the primary user-facing API in Spark since its inception. At the core, an RDD is an immutable distributed collection of elements of your data, partitioned across nodes in your cluster that can be operated in parallel with a low-level API that offers transformations and actions.\n",
    "\n",
    "### When to use RDDs?\n",
    "Consider these scenarios or common use cases for using RDDs when:\n",
    "- you want low-level transformation and actions and control on your dataset;\n",
    "- your data is unstructured, such as media streams or streams of text;\n",
    "- you want to manipulate your data with functional programming constructs than domain specific expressions;\n",
    "- you don’t care about imposing a schema, such as columnar format, while processing or accessing data attributes by name or column; and\n",
    "- you can forgo some optimization and performance benefits available with DataFrames and Datasets for structured and semi-structured data.\n",
    "\n",
    "### What happens to RDDs in Apache Spark 2.0?\n",
    "You may ask: Are RDDs being relegated as second class citizens? Are they being deprecated?\n",
    "\n",
    "**The answer is a resounding NO!**\n",
    "\n",
    "What’s more, as you will note below, you can seamlessly move between DataFrame or Dataset and RDDs at will—by simple API method calls—and DataFrames and Datasets are built on top of RDDs."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### DataFrames\n",
    "\n",
    "Like an RDD, a DataFrame is an immutable distributed collection of data. Unlike an RDD, data is organized into named columns, like a table in a relational database. Designed to make large data sets processing even easier, DataFrame allows developers to impose a structure onto a distributed collection of data, allowing higher-level abstraction; it provides a domain specific language API to manipulate your distributed data; and makes Spark accessible to a wider audience, beyond specialized data engineers.\n",
    "\n",
    "In Spark 2.0, DataFrame APIs will merge with Datasets APIs, unifying data processing capabilities across libraries. Because of this unification, developers now have fewer concepts to learn or remember, and work with a single high-level and type-safe API called `Dataset`.\n",
    "\n",
    "![Spark](https://databricks.com/wp-content/uploads/2016/06/Unified-Apache-Spark-2.0-API-1.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Datasets\n",
    "Starting in Spark 2.0, Dataset takes on two distinct APIs characteristics: a strongly-typed API and an untyped API, as shown in the table below. Conceptually, consider DataFrame as an alias for a collection of generic objects Dataset[Row], where a Row is a generic untyped JVM object. Dataset, by contrast, is a collection of strongly-typed JVM objects, dictated by a case class you define in Scala or a class in Java."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Typed and Un-typed APIs\n",
    "\n",
    "<table class=\"table\">\n",
    "<thead>\n",
    "<tr>\n",
    "<th>Language</th>\n",
    "<th>Main Abstraction</th>\n",
    "</tr>\n",
    "</thead>\n",
    "<tbody>\n",
    "<tr>\n",
    "<td>Scala</td>\n",
    "<td>Dataset[T] &amp; DataFrame (alias for Dataset[Row])</td>\n",
    "</tr>\n",
    "<tr>\n",
    "<td>Java</td>\n",
    "<td>Dataset[T]</td>\n",
    "</tr>\n",
    "<tr>\n",
    "<td>Python*</td>\n",
    "<td>DataFrame</td>\n",
    "</tr>\n",
    "<tr>\n",
    "<td>R*</td>\n",
    "<td>DataFrame</td>\n",
    "</tr>\n",
    "</tbody>\n",
    "</table>\n",
    "\n",
    "> **Note:** *Since Python and R have no compile-time type-safety, we only have untyped APIs, namely DataFrames.*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Benefits of Dataset APIs\n",
    "\n",
    "### 1. Static-typing and runtime type-safety\n",
    "\n",
    "Consider static-typing and runtime safety as a spectrum, with SQL least restrictive to Dataset most restrictive. For instance, in your Spark SQL string queries, you won’t know a syntax error until runtime (which could be costly), whereas in DataFrames and Datasets you can catch errors at compile time (which saves developer-time and costs). That is, if you invoke a function in DataFrame that is not part of the API, the compiler will catch it. However, it won’t detect a non-existing column name until runtime.\n",
    "\n",
    "At the far end of the spectrum is Dataset, most restrictive. Since Dataset APIs are all expressed as lambda functions and JVM typed objects, any mismatch of typed-parameters will be detected at compile time. Also, your analysis error can be detected at compile time too, when using Datasets, hence saving developer-time and costs.\n",
    "\n",
    "All this translates to is a spectrum of type-safety along syntax and analysis error in your Spark code, with Datasets as most restrictive yet productive for a developer.\n",
    "\n",
    "![](https://databricks.com/wp-content/uploads/2016/07/sql-vs-dataframes-vs-datasets-type-safety-spectrum.png)\n",
    "\n",
    "### 2. High-level abstraction and custom view into structured and semi-structured data\n",
    "DataFrames as a collection of Datasets[Row] render a structured custom view into your semi-structured data.\n",
    "\n",
    "### 3. Ease-of-use of APIs with structure\n",
    "\n",
    "Although structure may limit control in what your Spark program can do with data, it introduces rich semantics and an easy set of domain specific operations that can be expressed as high-level constructs. Most computations, however, can be accomplished with Dataset’s high-level APIs. For example, it’s much simpler to perform `agg`, `select`, `sum`, `avg`, `map`, `filter`, or `groupBy` operations. \n",
    "\n",
    "### 4. Performance and Optimization\n",
    "Along with all the above benefits, you cannot overlook the space efficiency and performance gains in using DataFrames and Dataset APIs for two reasons.\n",
    "\n",
    "First, because DataFrame and Dataset APIs are built on top of the Spark SQL engine, it uses Catalyst to generate an optimized logical and physical query plan. Across R, Java, Scala, or Python DataFrame/Dataset APIs, all relation type queries undergo the same code optimizer, providing the space and speed efficiency. Whereas the Dataset[T] typed API is optimized for data engineering tasks, the untyped Dataset[Row] (an alias of DataFrame) is even faster and suitable for interactive analysis.\n",
    "\n",
    "![](https://databricks.com/wp-content/uploads/2016/07/memory-usage-when-caching-datasets-vs-rdds.png)\n",
    "\n",
    "Second, since Spark as a compiler understands your Dataset type JVM object, it maps your type-specific JVM object to Tungsten’s internal memory representation using Encoders. As a result, Tungsten Encoders can efficiently serialize/deserialize JVM objects as well as generate compact bytecode that can execute at superior speeds.\n",
    "\n",
    "### When should I use DataFrames or Datasets?\n",
    "- If you want rich semantics, high-level abstractions, and domain specific APIs, use DataFrame or Dataset.\n",
    "- If your processing demands high-level expressions, filters, maps, aggregation, averages, sum, SQL queries, columnar access and use of lambda functions on semi-structured data, use DataFrame or Dataset.\n",
    "- If you want higher degree of type-safety at compile time, want typed JVM objects, take advantage of Catalyst optimization, and benefit from Tungsten’s efficient code generation, use Dataset.\n",
    "- If you want unification and simplification of APIs across Spark Libraries, use DataFrame or Dataset.\n",
    "- If you are a R user, use DataFrames.\n",
    "- If you are a Python user, use DataFrames and resort back to RDDs if you need more control."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Starting a Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "# May take a little while on a local computer\n",
    "spark = SparkSession.builder.appName(\"Structured API\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2020-07-04 15:35:17--  https://raw.githubusercontent.com/databricks/Spark-The-Definitive-Guide/master/data/flight-data/json/2015-summary.json\n",
      "Resolving raw.githubusercontent.com (raw.githubusercontent.com)... 151.101.8.133\n",
      "Connecting to raw.githubusercontent.com (raw.githubusercontent.com)|151.101.8.133|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 21368 (21K) [text/plain]\n",
      "Saving to: ‘data/2015-summary.json.2’\n",
      "\n",
      "2015-summary.json.2 100%[===================>]  20.87K  94.7KB/s    in 0.2s    \n",
      "\n",
      "2020-07-04 15:35:19 (94.7 KB/s) - ‘data/2015-summary.json.2’ saved [21368/21368]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!wget https://raw.githubusercontent.com/databricks/Spark-The-Definitive-Guide/master/data/flight-data/json/2015-summary.json -P data/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.format(\"json\").load(\"data/2015-summary.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- DEST_COUNTRY_NAME: string (nullable = true)\n",
      " |-- ORIGIN_COUNTRY_NAME: string (nullable = true)\n",
      " |-- count: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType(List(StructField(DEST_COUNTRY_NAME,StringType,true),StructField(ORIGIN_COUNTRY_NAME,StringType,true),StructField(count,LongType,true)))"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.read.format(\"json\").load(\"data/2015-summary.json\").schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*A schema is a StructType made up of a number of fields, StructFields, that have a name, type, a Boolean flag which specifies whether that column can contain missing or null values, and, finally, users can optionally specify associated metadata with that column. The metadata is a way of storing information about this column.*\n",
    "\n",
    "*If the types in the data (at runtime) do not match the schema, Spark will throw an error. The example that follows shows how to create and enforce a specific schema on a DataFrame.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructField, StructType, StringType, LongType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "myManualSchema = StructType([StructField(\"DEST_COUNTRY_NAME\", StringType(), True),\n",
    "                             StructField(\"ORIGIN_COUNTRY_NAME\", StringType(), True),\n",
    "                             StructField(\"count\", LongType(), False, metadata={\"hello\":\"world\"}) \n",
    "                            ])\n",
    "df = spark.read.format(\"json\").schema(myManualSchema).load(\"data/2015-summary.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- DEST_COUNTRY_NAME: string (nullable = true)\n",
      " |-- ORIGIN_COUNTRY_NAME: string (nullable = true)\n",
      " |-- count: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Spark Types**\n",
    "\n",
    "<table class=\"table\">\n",
    "<tbody><tr>\n",
    "  <th style=\"width:20%\">Data type</th>\n",
    "  <th style=\"width:40%\">Value type in Python</th>\n",
    "  <th>API to access or create a data type</th></tr>\n",
    "<tr>\n",
    "  <td> <b>ByteType</b> </td>\n",
    "  <td>\n",
    "  int or long <br>\n",
    "  <b>Note:</b> Numbers will be converted to 1-byte signed integer numbers at runtime.\n",
    "  Please make sure that numbers are within the range of -128 to 127.\n",
    "  </td>\n",
    "  <td>\n",
    "  ByteType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>ShortType</b> </td>\n",
    "  <td>\n",
    "  int or long <br>\n",
    "  <b>Note:</b> Numbers will be converted to 2-byte signed integer numbers at runtime.\n",
    "  Please make sure that numbers are within the range of -32768 to 32767.\n",
    "  </td>\n",
    "  <td>\n",
    "  ShortType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>IntegerType</b> </td>\n",
    "  <td> int or long </td>\n",
    "  <td>\n",
    "  IntegerType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>LongType</b> </td>\n",
    "  <td>\n",
    "  long <br>\n",
    "  <b>Note:</b> Numbers will be converted to 8-byte signed integer numbers at runtime.\n",
    "  Please make sure that numbers are within the range of\n",
    "  -9223372036854775808 to 9223372036854775807.\n",
    "  Otherwise, please convert data to decimal.Decimal and use DecimalType.\n",
    "  </td>\n",
    "  <td>\n",
    "  LongType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>FloatType</b> </td>\n",
    "  <td>\n",
    "  float <br>\n",
    "  <b>Note:</b> Numbers will be converted to 4-byte single-precision floating\n",
    "  point numbers at runtime.\n",
    "  </td>\n",
    "  <td>\n",
    "  FloatType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>DoubleType</b> </td>\n",
    "  <td> float </td>\n",
    "  <td>\n",
    "  DoubleType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>DecimalType</b> </td>\n",
    "  <td> decimal.Decimal </td>\n",
    "  <td>\n",
    "  DecimalType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>StringType</b> </td>\n",
    "  <td> string </td>\n",
    "  <td>\n",
    "  StringType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>BinaryType</b> </td>\n",
    "  <td> bytearray </td>\n",
    "  <td>\n",
    "  BinaryType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>BooleanType</b> </td>\n",
    "  <td> bool </td>\n",
    "  <td>\n",
    "  BooleanType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>TimestampType</b> </td>\n",
    "  <td> datetime.datetime </td>\n",
    "  <td>\n",
    "  TimestampType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>DateType</b> </td>\n",
    "  <td> datetime.date </td>\n",
    "  <td>\n",
    "  DateType()\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>ArrayType</b> </td>\n",
    "  <td> list, tuple, or array </td>\n",
    "  <td>\n",
    "  ArrayType(<i>elementType</i>, [<i>containsNull</i>])<br>\n",
    "  <b>Note:</b> The default value of <i>containsNull</i> is <i>True</i>.\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>MapType</b> </td>\n",
    "  <td> dict </td>\n",
    "  <td>\n",
    "  MapType(<i>keyType</i>, <i>valueType</i>, [<i>valueContainsNull</i>])<br>\n",
    "  <b>Note:</b> The default value of <i>valueContainsNull</i> is <i>True</i>.\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>StructType</b> </td>\n",
    "  <td> list or tuple </td>\n",
    "  <td>\n",
    "  StructType(<i>fields</i>)<br>\n",
    "  <b>Note:</b> <i>fields</i> is a Seq of StructFields. Also, two fields with the same\n",
    "  name are not allowed.\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>StructField</b> </td>\n",
    "  <td> The value type in Python of the data type of this field\n",
    "  (For example, Int for a StructField with the data type IntegerType) </td>\n",
    "  <td>\n",
    "  StructField(<i>name</i>, <i>dataType</i>, [<i>nullable</i>])<br>\n",
    "  <b>Note:</b> The default value of <i>nullable</i> is <i>True</i>.\n",
    "  </td>\n",
    "</tr>\n",
    "</tbody></table>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------------------+-----+\n",
      "|   DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+--------------------+-------------------+-----+\n",
      "|       United States|            Romania|   15|\n",
      "|       United States|            Croatia|    1|\n",
      "|       United States|            Ireland|  344|\n",
      "|               Egypt|      United States|   15|\n",
      "|       United States|              India|   62|\n",
      "|       United States|          Singapore|    1|\n",
      "|       United States|            Grenada|   62|\n",
      "|          Costa Rica|      United States|  588|\n",
      "|             Senegal|      United States|   40|\n",
      "|             Moldova|      United States|    1|\n",
      "|       United States|       Sint Maarten|  325|\n",
      "|       United States|   Marshall Islands|   39|\n",
      "|              Guyana|      United States|   64|\n",
      "|               Malta|      United States|    1|\n",
      "|            Anguilla|      United States|   41|\n",
      "|             Bolivia|      United States|   30|\n",
      "|       United States|           Paraguay|    6|\n",
      "|             Algeria|      United States|    4|\n",
      "|Turks and Caicos ...|      United States|  230|\n",
      "|       United States|          Gibraltar|    1|\n",
      "+--------------------+-------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Romania', count=15)"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*This is a row object in spark DataFrame*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'United States'"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.first().asDict()['DEST_COUNTRY_NAME']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Romania', count=15),\n",
       " Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Croatia', count=1),\n",
       " Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Ireland', count=344)]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Columns and Expressions\n",
    "\n",
    "There are a lot of different ways to construct and refer to columns but the two simplest ways are\n",
    "by using the `col` or `column` functions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col, column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|DEST_COUNTRY_NAME|\n",
      "+-----------------+\n",
      "|    United States|\n",
      "|    United States|\n",
      "+-----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(col(\"DEST_COUNTRY_NAME\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|DEST_COUNTRY_NAME|\n",
      "+-----------------+\n",
      "|    United States|\n",
      "|    United States|\n",
      "+-----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(column(\"DEST_COUNTRY_NAME\")).show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Expressions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|         Destination|\n",
      "+--------------------+\n",
      "|       United States|\n",
      "|       United States|\n",
      "|       United States|\n",
      "|               Egypt|\n",
      "|       United States|\n",
      "|       United States|\n",
      "|       United States|\n",
      "|          Costa Rica|\n",
      "|             Senegal|\n",
      "|             Moldova|\n",
      "|       United States|\n",
      "|       United States|\n",
      "|              Guyana|\n",
      "|               Malta|\n",
      "|            Anguilla|\n",
      "|             Bolivia|\n",
      "|       United States|\n",
      "|             Algeria|\n",
      "|Turks and Caicos ...|\n",
      "|       United States|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import expr\n",
    "df.select(expr(\"DEST_COUNTRY_NAME as Destination\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## select and selectExpr\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|DEST_COUNTRY_NAME|\n",
      "+-----------------+\n",
      "|    United States|\n",
      "|    United States|\n",
      "+-----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"DEST_COUNTRY_NAME\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|\n",
      "+-----------------+-------------------+\n",
      "|    United States|            Romania|\n",
      "|    United States|            Croatia|\n",
      "+-----------------+-------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"DEST_COUNTRY_NAME\", \"ORIGIN_COUNTRY_NAME\").show(2)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-----------------+-----------------+\n",
      "|DEST_COUNTRY_NAME|DEST_COUNTRY_NAME|DEST_COUNTRY_NAME|\n",
      "+-----------------+-----------------+-----------------+\n",
      "|    United States|    United States|    United States|\n",
      "|    United States|    United States|    United States|\n",
      "+-----------------+-----------------+-----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(expr(\"DEST_COUNTRY_NAME\"),\n",
    "                col(\"DEST_COUNTRY_NAME\"),\n",
    "                column(\"DEST_COUNTRY_NAME\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+\n",
      "|  destination|\n",
      "+-------------+\n",
      "|United States|\n",
      "|United States|\n",
      "+-------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(expr(\"DEST_COUNTRY_NAME AS destination\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|DEST_COUNTRY_NAME|\n",
      "+-----------------+\n",
      "|    United States|\n",
      "|    United States|\n",
      "+-----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(expr(\"DEST_COUNTRY_NAME as destination\").alias(\"DEST_COUNTRY_NAME\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-----------------+\n",
      "|newColumnName|DEST_COUNTRY_NAME|\n",
      "+-------------+-----------------+\n",
      "|United States|    United States|\n",
      "|United States|    United States|\n",
      "+-------------+-----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.selectExpr(\"DEST_COUNTRY_NAME as newColumnName\", \"DEST_COUNTRY_NAME\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+-------------+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|withinCountry|\n",
      "+-----------------+-------------------+-----+-------------+\n",
      "|    United States|            Romania|   15|        false|\n",
      "|    United States|            Croatia|    1|        false|\n",
      "+-----------------+-------------------+-----+-------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.selectExpr(\n",
    "\"*\", # all original columns\n",
    "\"(DEST_COUNTRY_NAME = ORIGIN_COUNTRY_NAME) as withinCountry\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---------------------------------+\n",
      "| avg(count)|count(DISTINCT DEST_COUNTRY_NAME)|\n",
      "+-----------+---------------------------------+\n",
      "|1770.765625|                              132|\n",
      "+-----------+---------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.selectExpr(\"avg(count)\", \"count(distinct(DEST_COUNTRY_NAME))\").show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Adding Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+---------+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|numberOne|\n",
      "+-----------------+-------------------+-----+---------+\n",
      "|    United States|            Romania|   15|        1|\n",
      "|    United States|            Croatia|    1|        1|\n",
      "+-----------------+-------------------+-----+---------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Adding a column with literal 1\n",
    "from pyspark.sql.functions import lit\n",
    "df.withColumn(\"numberOne\", lit(1)).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+-------------+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|withinCountry|\n",
      "+-----------------+-------------------+-----+-------------+\n",
      "|    United States|            Romania|   15|        false|\n",
      "|    United States|            Croatia|    1|        false|\n",
      "+-----------------+-------------------+-----+-------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn(\"withinCountry\", expr(\"ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['dest', 'ORIGIN_COUNTRY_NAME', 'count']"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Renaming Columns\n",
    "df.withColumnRenamed(\"DEST_COUNTRY_NAME\", \"dest\").columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+-------------+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|withinCountry|\n",
      "+-----------------+-------------------+-----+-------------+\n",
      "|    United States|            Romania|   15|        false|\n",
      "|    United States|            Croatia|    1|        false|\n",
      "+-----------------+-------------------+-----+-------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn(\"withinCountry\", expr(\"ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['DEST_COUNTRY_NAME', 'count']"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Removing Columns\n",
    "df.drop(\"ORIGIN_COUNTRY_NAME\").columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType(List(StructField(DEST_COUNTRY_NAME,StringType,true),StructField(ORIGIN_COUNTRY_NAME,StringType,true),StructField(count,LongType,true),StructField(count2,LongType,true)))"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Changing a Column’s Type (cast)\n",
    "df.withColumn(\"count2\", col(\"count\").cast(\"long\")).schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Filtering Rows\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+-----------------+-------------------+-----+\n",
      "|    United States|            Croatia|    1|\n",
      "|    United States|          Singapore|    1|\n",
      "+-----------------+-------------------+-----+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(col(\"count\") < 2).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+-----------------+-------------------+-----+\n",
      "|    United States|            Croatia|    1|\n",
      "|    United States|          Singapore|    1|\n",
      "+-----------------+-------------------+-----+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.where(\"count < 2\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+-----------------+-------------------+-----+\n",
      "|    United States|          Singapore|    1|\n",
      "|          Moldova|      United States|    1|\n",
      "+-----------------+-------------------+-----+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.where(col(\"count\") < 2).where(col(\"ORIGIN_COUNTRY_NAME\") != \"Croatia\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "256"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Getting Unique Rows\n",
    "df.select(\"ORIGIN_COUNTRY_NAME\", \"DEST_COUNTRY_NAME\").distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "125"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select(\"ORIGIN_COUNTRY_NAME\").distinct().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Random Samples\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "138"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "seed = 5\n",
    "withReplacement = False\n",
    "fraction = 0.5\n",
    "df.sample(withReplacement, fraction, seed).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Concatenating and Appending Rows (Union)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Row\n",
    "schema = df.schema\n",
    "newRows = [Row(\"New Country\", \"Other Country\", 5),Row(\"New Country 2\", \"Other Country 3\", 1)]\n",
    "parallelizedRows = spark.sparkContext.parallelize(newRows)\n",
    "newDF = spark.createDataFrame(parallelizedRows, schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+-----------------+-------------------+-----+\n",
      "|    United States|            Croatia|    1|\n",
      "|    United States|          Singapore|    1|\n",
      "|    United States|          Gibraltar|    1|\n",
      "|    United States|             Cyprus|    1|\n",
      "|    United States|            Estonia|    1|\n",
      "|    United States|          Lithuania|    1|\n",
      "|    United States|           Bulgaria|    1|\n",
      "|    United States|            Georgia|    1|\n",
      "|    United States|            Bahrain|    1|\n",
      "|    United States|   Papua New Guinea|    1|\n",
      "|    United States|         Montenegro|    1|\n",
      "|    United States|            Namibia|    1|\n",
      "|    New Country 2|    Other Country 3|    1|\n",
      "+-----------------+-------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.union(newDF).where(\"count = 1\").where(col(\"ORIGIN_COUNTRY_NAME\") != \"United States\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sorting Rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------------------+-----+\n",
      "|   DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+--------------------+-------------------+-----+\n",
      "|               Malta|      United States|    1|\n",
      "|Saint Vincent and...|      United States|    1|\n",
      "|       United States|            Croatia|    1|\n",
      "|       United States|          Gibraltar|    1|\n",
      "|       United States|          Singapore|    1|\n",
      "+--------------------+-------------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.sort(\"count\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+-----------------+-------------------+-----+\n",
      "|     Burkina Faso|      United States|    1|\n",
      "|    Cote d'Ivoire|      United States|    1|\n",
      "|           Cyprus|      United States|    1|\n",
      "|         Djibouti|      United States|    1|\n",
      "|        Indonesia|      United States|    1|\n",
      "+-----------------+-------------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.orderBy(\"count\", \"DEST_COUNTRY_NAME\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+-----------------+-------------------+-----+\n",
      "|     Burkina Faso|      United States|    1|\n",
      "|    Cote d'Ivoire|      United States|    1|\n",
      "|           Cyprus|      United States|    1|\n",
      "|         Djibouti|      United States|    1|\n",
      "|        Indonesia|      United States|    1|\n",
      "+-----------------+-------------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.orderBy(col(\"count\"), col(\"DEST_COUNTRY_NAME\")).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*By default it sorts in ascending order but if you want to explicitly define the order use desc and asc*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+-----+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+-----------------+-------------------+-----+\n",
      "|          Moldova|      United States|    1|\n",
      "|    United States|            Croatia|    1|\n",
      "+-----------------+-------------------+-----+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import desc, asc\n",
    "df.orderBy(expr(\"count desc\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-------------------+------+\n",
      "|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME| count|\n",
      "+-----------------+-------------------+------+\n",
      "|    United States|      United States|370002|\n",
      "|    United States|             Canada|  8483|\n",
      "+-----------------+-------------------+------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.orderBy(col(\"count\").desc(), col(\"DEST_COUNTRY_NAME\").asc()).show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Limit"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------------------+-----+\n",
      "|   DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|\n",
      "+--------------------+-------------------+-----+\n",
      "|               Malta|      United States|    1|\n",
      "|Saint Vincent and...|      United States|    1|\n",
      "|       United States|            Croatia|    1|\n",
      "|       United States|          Gibraltar|    1|\n",
      "|       United States|          Singapore|    1|\n",
      "|             Moldova|      United States|    1|\n",
      "+--------------------+-------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.orderBy(expr(\"count desc\")).limit(6).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Repartition and Coalesce"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string, count: bigint]"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.repartition(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.repartition(5).rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string, count: bigint]"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# If you know that you’re going to be filtering by a certain column often, \n",
    "# it can be worth repartitioning based on that column\n",
    "df.repartition(col(\"DEST_COUNTRY_NAME\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string, count: bigint]"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.repartition(5, col(\"DEST_COUNTRY_NAME\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Coalesce, on the other hand, will not incur a full shuffle and will try to combine partitions.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string, count: bigint]"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.repartition(5, col(\"DEST_COUNTRY_NAME\")).coalesce(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Working with Different Types of Data\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2020-07-04 15:47:30--  https://raw.githubusercontent.com/databricks/Spark-The-Definitive-Guide/master/data/retail-data/by-day/2010-12-01.csv\n",
      "Resolving raw.githubusercontent.com (raw.githubusercontent.com)... 151.101.8.133\n",
      "Connecting to raw.githubusercontent.com (raw.githubusercontent.com)|151.101.8.133|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 275001 (269K) [text/plain]\n",
      "Saving to: ‘data/2010-12-01.csv’\n",
      "\n",
      "2010-12-01.csv      100%[===================>] 268.56K   350KB/s    in 0.8s    \n",
      "\n",
      "2020-07-04 15:47:32 (350 KB/s) - ‘data/2010-12-01.csv’ saved [275001/275001]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!wget https://raw.githubusercontent.com/databricks/Spark-The-Definitive-Guide/master/data/retail-data/by-day/2010-12-01.csv -P data/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- InvoiceNo: string (nullable = true)\n",
      " |-- StockCode: string (nullable = true)\n",
      " |-- Description: string (nullable = true)\n",
      " |-- Quantity: integer (nullable = true)\n",
      " |-- InvoiceDate: string (nullable = true)\n",
      " |-- UnitPrice: double (nullable = true)\n",
      " |-- CustomerID: double (nullable = true)\n",
      " |-- Country: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.read.format(\"csv\")\\\n",
    ".option(\"header\", \"true\")\\\n",
    ".option(\"inferSchema\", \"true\")\\\n",
    ".load(\"data/2010-12-01.csv\")\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[5: int, five: string, 5.0: double]"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import lit\n",
    "df.select(lit(5), lit(\"five\"), lit(5.0))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Working with Booleans\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-----------------------------+\n",
      "|InvoiceNo|Description                  |\n",
      "+---------+-----------------------------+\n",
      "|536366   |HAND WARMER UNION JACK       |\n",
      "|536366   |HAND WARMER RED POLKA DOT    |\n",
      "|536367   |ASSORTED COLOUR BIRD ORNAMENT|\n",
      "|536367   |POPPY'S PLAYHOUSE BEDROOM    |\n",
      "|536367   |POPPY'S PLAYHOUSE KITCHEN    |\n",
      "+---------+-----------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import col\n",
    "df.where(col(\"InvoiceNo\") != 536365)\\\n",
    ".select(\"InvoiceNo\", \"Description\")\\\n",
    ".show(5, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+--------------+--------+-------------------+---------+----------+--------------+\n",
      "|InvoiceNo|StockCode|   Description|Quantity|        InvoiceDate|UnitPrice|CustomerID|       Country|\n",
      "+---------+---------+--------------+--------+-------------------+---------+----------+--------------+\n",
      "|   536544|      DOT|DOTCOM POSTAGE|       1|2010-12-01 14:32:00|   569.77|      null|United Kingdom|\n",
      "|   536592|      DOT|DOTCOM POSTAGE|       1|2010-12-01 17:06:00|   607.49|      null|United Kingdom|\n",
      "+---------+---------+--------------+--------+-------------------+---------+----------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import instr\n",
    "priceFilter = col(\"UnitPrice\") > 600\n",
    "descripFilter = instr(df.Description, \"POSTAGE\") >= 1\n",
    "df.where(df.StockCode.isin(\"DOT\")).where(priceFilter | descripFilter).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-----------+\n",
      "|unitPrice|isExpensive|\n",
      "+---------+-----------+\n",
      "|   569.77|       true|\n",
      "|   607.49|       true|\n",
      "+---------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "DOTCodeFilter = col(\"StockCode\") == \"DOT\"\n",
    "priceFilter = col(\"UnitPrice\") > 600\n",
    "descripFilter = instr(col(\"Description\"), \"POSTAGE\") >= 1\n",
    "df.withColumn(\"isExpensive\", DOTCodeFilter & (priceFilter | descripFilter))\\\n",
    ".where(\"isExpensive\")\\\n",
    ".select(\"unitPrice\", \"isExpensive\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---------+\n",
      "|   Description|UnitPrice|\n",
      "+--------------+---------+\n",
      "|DOTCOM POSTAGE|   569.77|\n",
      "|DOTCOM POSTAGE|   607.49|\n",
      "+--------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn(\"isExpensive\", expr(\"NOT UnitPrice <= 250\"))\\\n",
    ".where(\"isExpensive\")\\\n",
    ".select(\"Description\", \"UnitPrice\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Working with Numbers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------------+\n",
      "|CustomerId|      realQuantity|\n",
      "+----------+------------------+\n",
      "|   17850.0|239.08999999999997|\n",
      "|   17850.0|          418.7156|\n",
      "+----------+------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import expr, pow\n",
    "fabricatedQuantity = pow(col(\"Quantity\") * col(\"UnitPrice\"), 2) + 5\n",
    "df.select(expr(\"CustomerId\"), fabricatedQuantity.alias(\"realQuantity\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------------+\n",
      "|CustomerId|      realQuantity|\n",
      "+----------+------------------+\n",
      "|   17850.0|239.08999999999997|\n",
      "|   17850.0|          418.7156|\n",
      "+----------+------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.selectExpr(\n",
    "\"CustomerId\",\n",
    "\"(POWER((Quantity * UnitPrice), 2.0) + 5) as realQuantity\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------------+\n",
      "|round(2.5, 0)|bround(2.5, 0)|\n",
      "+-------------+--------------+\n",
      "|          3.0|           2.0|\n",
      "|          3.0|           2.0|\n",
      "+-------------+--------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import lit, round, bround\n",
    "df.select(round(lit(\"2.5\")), bround(lit(\"2.5\"))).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+\n",
      "|corr(Quantity, UnitPrice)|\n",
      "+-------------------------+\n",
      "|     -0.04112314436835551|\n",
      "+-------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import corr\n",
    "df.stat.corr(\"Quantity\", \"UnitPrice\")\n",
    "df.select(corr(\"Quantity\", \"UnitPrice\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+------------------+--------------------+------------------+-------------------+------------------+------------------+--------------+\n",
      "|summary|        InvoiceNo|         StockCode|         Description|          Quantity|        InvoiceDate|         UnitPrice|        CustomerID|       Country|\n",
      "+-------+-----------------+------------------+--------------------+------------------+-------------------+------------------+------------------+--------------+\n",
      "|  count|             3108|              3108|                3098|              3108|               3108|              3108|              1968|          3108|\n",
      "|   mean| 536516.684944841|27834.304044117645|                null| 8.627413127413128|               null| 4.151946589446603|15661.388719512195|          null|\n",
      "| stddev|72.89447869788873|17407.897548583845|                null|26.371821677029203|               null|15.638659854603892|1854.4496996893627|          null|\n",
      "|    min|           536365|             10002| 4 PURPLE FLOCK D...|               -24|2010-12-01 08:26:00|               0.0|           12431.0|     Australia|\n",
      "|    max|          C536548|              POST|ZINC WILLIE WINKI...|               600|2010-12-01 17:35:00|            607.49|           18229.0|United Kingdom|\n",
      "+-------+-----------------+------------------+--------------------+------------------+-------------------+------------------+------------------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[2.51]"
      ]
     },
     "execution_count": 72,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import count, mean, stddev_pop, min, max\n",
    "colName = \"UnitPrice\"\n",
    "quantileProbs = [0.5]\n",
    "relError = 0.05\n",
    "df.stat.approxQuantile(\"UnitPrice\", quantileProbs, relError) # 2.51"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+\n",
      "|StockCode_Quantity| -1|-10|-12| -2|-24| -3| -4| -5| -6| -7|  1| 10|100| 11| 12|120|128| 13| 14|144| 15| 16| 17| 18| 19|192|  2| 20|200| 21|216| 22| 23| 24| 25|252| 27| 28|288|  3| 30| 32| 33| 34| 36|384|  4| 40|432| 47| 48|480|  5| 50| 56|  6| 60|600| 64|  7| 70| 72|  8| 80|  9| 96|\n",
      "+------------------+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+\n",
      "|             22578|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             21327|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  2|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             22064|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             21080|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|\n",
      "|             22219|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  3|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             21908|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             22818|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|           15056BL|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             72817|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             22545|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             22988|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|\n",
      "|             22274|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             20750|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  2|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|            82616C|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             21703|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             22899|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  2|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|\n",
      "|             22379|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  2|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             22422|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  2|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             22769|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "|             22585|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|  1|  0|  0|  1|  0|  0|  0|  0|  0|  0|  0|  0|  0|  0|\n",
      "+------------------+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.stat.crosstab(\"StockCode\", \"Quantity\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "| StockCode_freqItems|  Quantity_freqItems|\n",
      "+--------------------+--------------------+\n",
      "|[90214E, 20728, 2...|[200, 128, 23, 32...|\n",
      "+--------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.stat.freqItems([\"StockCode\", \"Quantity\"]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------+\n",
      "|monotonically_increasing_id()|\n",
      "+-----------------------------+\n",
      "|                            0|\n",
      "|                            1|\n",
      "+-----------------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import monotonically_increasing_id\n",
    "df.select(monotonically_increasing_id()).show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Working with Strings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|initcap(Description)|\n",
      "+--------------------+\n",
      "|White Hanging Hea...|\n",
      "| White Metal Lantern|\n",
      "|Cream Cupid Heart...|\n",
      "|Knitted Union Fla...|\n",
      "|Red Woolly Hottie...|\n",
      "|Set 7 Babushka Ne...|\n",
      "|Glass Star Froste...|\n",
      "|Hand Warmer Union...|\n",
      "|Hand Warmer Red P...|\n",
      "|Assorted Colour B...|\n",
      "|Poppy's Playhouse...|\n",
      "|Poppy's Playhouse...|\n",
      "|Feltcraft Princes...|\n",
      "|Ivory Knitted Mug...|\n",
      "|Box Of 6 Assorted...|\n",
      "|Box Of Vintage Ji...|\n",
      "|Box Of Vintage Al...|\n",
      "|Home Building Blo...|\n",
      "|Love Building Blo...|\n",
      "|Recipe Box With M...|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import initcap\n",
    "df.select(initcap(col(\"Description\"))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-------------------------+\n",
      "|         Description|  lower(Description)|upper(lower(Description))|\n",
      "+--------------------+--------------------+-------------------------+\n",
      "|WHITE HANGING HEA...|white hanging hea...|     WHITE HANGING HEA...|\n",
      "| WHITE METAL LANTERN| white metal lantern|      WHITE METAL LANTERN|\n",
      "+--------------------+--------------------+-------------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import lower, upper\n",
    "df.select(col(\"Description\"),\n",
    "lower(col(\"Description\")),\n",
    "upper(lower(col(\"Description\")))).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------+-----+---+----------+\n",
      "|   ltrim|   rtrim| trim| lp|        rp|\n",
      "+--------+--------+-----+---+----------+\n",
      "|HELLO   |   HELLO|HELLO|HEL|HELLO     |\n",
      "|HELLO   |   HELLO|HELLO|HEL|HELLO     |\n",
      "+--------+--------+-----+---+----------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import lit, ltrim, rtrim, rpad, lpad, trim\n",
    "df.select(\n",
    "ltrim(lit(\"   HELLO   \")).alias(\"ltrim\"),\n",
    "rtrim(lit(\"   HELLO   \")).alias(\"rtrim\"),\n",
    "trim(lit(\"    HELLO   \")).alias(\"trim\"),\n",
    "lpad(lit(\"HELLO\"), 3, \" \").alias(\"lp\"),\n",
    "rpad(lit(\"HELLO\"), 10, \" \").alias(\"rp\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "|         color_clean|         Description|\n",
      "+--------------------+--------------------+\n",
      "|COLOR HANGING HEA...|WHITE HANGING HEA...|\n",
      "| COLOR METAL LANTERN| WHITE METAL LANTERN|\n",
      "+--------------------+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Regular Expressions\n",
    "from pyspark.sql.functions import regexp_replace\n",
    "regex_string = \"BLACK|WHITE|RED|GREEN|BLUE\"\n",
    "df.select(\n",
    "regexp_replace(col(\"Description\"), regex_string, \"COLOR\").alias(\"color_clean\"),\n",
    "col(\"Description\")).show(2)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------+--------------------+\n",
      "|translate(Description, LEET, 1337)|         Description|\n",
      "+----------------------------------+--------------------+\n",
      "|              WHI73 HANGING H3A...|WHITE HANGING HEA...|\n",
      "|               WHI73 M37A1 1AN73RN| WHITE METAL LANTERN|\n",
      "+----------------------------------+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import translate\n",
    "df.select(translate(col(\"Description\"), \"LEET\", \"1337\"),col(\"Description\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------------------+\n",
      "|color_clean|         Description|\n",
      "+-----------+--------------------+\n",
      "|      WHITE|WHITE HANGING HEA...|\n",
      "|      WHITE| WHITE METAL LANTERN|\n",
      "+-----------+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import regexp_extract\n",
    "extract_str = \"(BLACK|WHITE|RED|GREEN|BLUE)\"\n",
    "df.select(\n",
    "regexp_extract(col(\"Description\"), extract_str, 1).alias(\"color_clean\"),\n",
    "col(\"Description\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------+\n",
      "|Description                       |\n",
      "+----------------------------------+\n",
      "|WHITE HANGING HEART T-LIGHT HOLDER|\n",
      "|WHITE METAL LANTERN               |\n",
      "|RED WOOLLY HOTTIE WHITE HEART.    |\n",
      "+----------------------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import instr\n",
    "containsBlack = instr(col(\"Description\"), \"BLACK\") >= 1\n",
    "containsWhite = instr(col(\"Description\"), \"WHITE\") >= 1\n",
    "df.withColumn(\"hasSimpleColor\", containsBlack | containsWhite)\\\n",
    ".where(\"hasSimpleColor\")\\\n",
    ".select(\"Description\").show(3, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Working with Dates and Timestamps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+--------------------+\n",
      "| id|     today|                 now|\n",
      "+---+----------+--------------------+\n",
      "|  0|2020-07-04|2020-07-04 16:14:...|\n",
      "|  1|2020-07-04|2020-07-04 16:14:...|\n",
      "|  2|2020-07-04|2020-07-04 16:14:...|\n",
      "|  3|2020-07-04|2020-07-04 16:14:...|\n",
      "|  4|2020-07-04|2020-07-04 16:14:...|\n",
      "|  5|2020-07-04|2020-07-04 16:14:...|\n",
      "|  6|2020-07-04|2020-07-04 16:14:...|\n",
      "|  7|2020-07-04|2020-07-04 16:14:...|\n",
      "|  8|2020-07-04|2020-07-04 16:14:...|\n",
      "|  9|2020-07-04|2020-07-04 16:14:...|\n",
      "+---+----------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import current_date, current_timestamp\n",
    "dateDF = spark.range(10)\\\n",
    ".withColumn(\"today\", current_date())\\\n",
    ".withColumn(\"now\", current_timestamp())\n",
    "dateDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: long (nullable = false)\n",
      " |-- today: date (nullable = false)\n",
      " |-- now: timestamp (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dateDF.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+------------------+\n",
      "|date_sub(today, 5)|date_add(today, 5)|\n",
      "+------------------+------------------+\n",
      "|        2020-06-29|        2020-07-09|\n",
      "+------------------+------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import date_add, date_sub\n",
    "dateDF.select(date_sub(col(\"today\"), 5), date_add(col(\"today\"), 5)).show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+\n",
      "|datediff(week_ago, today)|\n",
      "+-------------------------+\n",
      "|                       -7|\n",
      "+-------------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import datediff, months_between, to_date\n",
    "\n",
    "dateDF.withColumn(\"week_ago\", date_sub(col(\"today\"), 7))\\\n",
    ".select(datediff(col(\"week_ago\"), col(\"today\"))).show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+\n",
      "|months_between(start, end, true)|\n",
      "+--------------------------------+\n",
      "|                    -16.67741935|\n",
      "+--------------------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dateDF.select(\n",
    "to_date(lit(\"2016-01-01\")).alias(\"start\"),\n",
    "to_date(lit(\"2017-05-22\")).alias(\"end\"))\\\n",
    ".select(months_between(col(\"start\"), col(\"end\"))).show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+\n",
      "|to_date(`date`)|\n",
      "+---------------+\n",
      "|     2017-01-01|\n",
      "+---------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import to_date, lit\n",
    "spark.range(5).withColumn(\"date\", lit(\"2017-01-01\"))\\\n",
    ".select(to_date(col(\"date\"))).show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+\n",
      "|      date|     date2|\n",
      "+----------+----------+\n",
      "|2017-11-12|2017-12-20|\n",
      "+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import to_date\n",
    "dateFormat = \"yyyy-dd-MM\"\n",
    "cleanDateDF = spark.range(1).select(\n",
    "to_date(lit(\"2017-12-11\"), dateFormat).alias(\"date\"),\n",
    "to_date(lit(\"2017-20-12\"), dateFormat).alias(\"date2\"))\n",
    "cleanDateDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------+\n",
      "|to_timestamp(`date`, 'yyyy-dd-MM')|\n",
      "+----------------------------------+\n",
      "|               2017-11-12 00:00:00|\n",
      "+----------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import to_timestamp\n",
    "cleanDateDF.select(to_timestamp(col(\"date\"), dateFormat)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Working with Nulls in Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------------+\n",
      "|coalesce(Description, CustomerId)|\n",
      "+---------------------------------+\n",
      "|             WHITE HANGING HEA...|\n",
      "|              WHITE METAL LANTERN|\n",
      "|             CREAM CUPID HEART...|\n",
      "|             KNITTED UNION FLA...|\n",
      "|             RED WOOLLY HOTTIE...|\n",
      "|             SET 7 BABUSHKA NE...|\n",
      "|             GLASS STAR FROSTE...|\n",
      "|             HAND WARMER UNION...|\n",
      "|             HAND WARMER RED P...|\n",
      "|             ASSORTED COLOUR B...|\n",
      "|             POPPY'S PLAYHOUSE...|\n",
      "|             POPPY'S PLAYHOUSE...|\n",
      "|             FELTCRAFT PRINCES...|\n",
      "|             IVORY KNITTED MUG...|\n",
      "|             BOX OF 6 ASSORTED...|\n",
      "|             BOX OF VINTAGE JI...|\n",
      "|             BOX OF VINTAGE AL...|\n",
      "|             HOME BUILDING BLO...|\n",
      "|             LOVE BUILDING BLO...|\n",
      "|             RECIPE BOX WITH M...|\n",
      "+---------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import coalesce\n",
    "df.select(coalesce(col(\"Description\"), col(\"CustomerId\"))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[InvoiceNo: string, StockCode: string, Description: string, Quantity: int, InvoiceDate: string, UnitPrice: double, CustomerID: double, Country: string]"
      ]
     },
     "execution_count": 93,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.na.drop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[InvoiceNo: string, StockCode: string, Description: string, Quantity: int, InvoiceDate: string, UnitPrice: double, CustomerID: double, Country: string]"
      ]
     },
     "execution_count": 94,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.na.drop(\"any\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[InvoiceNo: string, StockCode: string, Description: string, Quantity: int, InvoiceDate: string, UnitPrice: double, CustomerID: double, Country: string]"
      ]
     },
     "execution_count": 95,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.na.drop(\"all\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[InvoiceNo: string, StockCode: string, Description: string, Quantity: int, InvoiceDate: string, UnitPrice: double, CustomerID: double, Country: string]"
      ]
     },
     "execution_count": 96,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.na.drop(\"all\", subset=[\"StockCode\", \"InvoiceNo\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[InvoiceNo: string, StockCode: string, Description: string, Quantity: int, InvoiceDate: string, UnitPrice: double, CustomerID: double, Country: string]"
      ]
     },
     "execution_count": 97,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.na.fill(\"all\", subset=[\"StockCode\", \"InvoiceNo\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[InvoiceNo: string, StockCode: string, Description: string, Quantity: int, InvoiceDate: string, UnitPrice: double, CustomerID: double, Country: string]"
      ]
     },
     "execution_count": 98,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.na.fill(\"All Null values become this string\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[InvoiceNo: string, StockCode: string, Description: string, Quantity: int, InvoiceDate: string, UnitPrice: double, CustomerID: double, Country: string]"
      ]
     },
     "execution_count": 99,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.na.replace([\"\"], [\"UNKNOWN\"], \"Description\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Working with Complex Types"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|             complex|\n",
      "+--------------------+\n",
      "|[WHITE HANGING HE...|\n",
      "|[WHITE METAL LANT...|\n",
      "|[CREAM CUPID HEAR...|\n",
      "|[KNITTED UNION FL...|\n",
      "|[RED WOOLLY HOTTI...|\n",
      "|[SET 7 BABUSHKA N...|\n",
      "|[GLASS STAR FROST...|\n",
      "|[HAND WARMER UNIO...|\n",
      "|[HAND WARMER RED ...|\n",
      "|[ASSORTED COLOUR ...|\n",
      "|[POPPY'S PLAYHOUS...|\n",
      "|[POPPY'S PLAYHOUS...|\n",
      "|[FELTCRAFT PRINCE...|\n",
      "|[IVORY KNITTED MU...|\n",
      "|[BOX OF 6 ASSORTE...|\n",
      "|[BOX OF VINTAGE J...|\n",
      "|[BOX OF VINTAGE A...|\n",
      "|[HOME BUILDING BL...|\n",
      "|[LOVE BUILDING BL...|\n",
      "|[RECIPE BOX WITH ...|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import struct\n",
    "complexDF = df.select(struct(\"Description\", \"InvoiceNo\").alias(\"complex\"))\n",
    "complexDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+\n",
      "|split(Description,  , -1)|\n",
      "+-------------------------+\n",
      "|     [WHITE, HANGING, ...|\n",
      "|     [WHITE, METAL, LA...|\n",
      "+-------------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import split\n",
    "df.select(split(col(\"Description\"), \" \")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+\n",
      "|array_col[0]|\n",
      "+------------+\n",
      "|       WHITE|\n",
      "|       WHITE|\n",
      "+------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(split(col(\"Description\"), \" \").alias(\"array_col\"))\\\n",
    ".selectExpr(\"array_col[0]\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------+\n",
      "|size(split(Description,  , -1))|\n",
      "+-------------------------------+\n",
      "|                              5|\n",
      "|                              3|\n",
      "+-------------------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import size\n",
    "df.select(size(split(col(\"Description\"), \" \"))).show(2) # shows 5 and 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------------------+\n",
      "|array_contains(split(Description,  , -1), WHITE)|\n",
      "+------------------------------------------------+\n",
      "|                                            true|\n",
      "|                                            true|\n",
      "+------------------------------------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import array_contains\n",
    "df.select(array_contains(split(col(\"Description\"), \" \"), \"WHITE\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------+--------+\n",
      "|         Description|InvoiceNo|exploded|\n",
      "+--------------------+---------+--------+\n",
      "|WHITE HANGING HEA...|   536365|   WHITE|\n",
      "|WHITE HANGING HEA...|   536365| HANGING|\n",
      "+--------------------+---------+--------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import split, explode\n",
    "df.withColumn(\"splitted\", split(col(\"Description\"), \" \"))\\\n",
    ".withColumn(\"exploded\", explode(col(\"splitted\")))\\\n",
    ".select(\"Description\", \"InvoiceNo\", \"exploded\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|         complex_map|\n",
      "+--------------------+\n",
      "|[WHITE HANGING HE...|\n",
      "|[WHITE METAL LANT...|\n",
      "+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Maps are created by using the map function and key-value pairs of columns\n",
    "from pyspark.sql.functions import create_map\n",
    "df.select(create_map(col(\"Description\"), col(\"InvoiceNo\")).alias(\"complex_map\")).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+\n",
      "|complex_map[WHITE METAL LANTERN]|\n",
      "+--------------------------------+\n",
      "|                            null|\n",
      "|                          536365|\n",
      "+--------------------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(create_map(col(\"Description\"), col(\"InvoiceNo\")).alias(\"complex_map\"))\\\n",
    ".selectExpr(\"complex_map['WHITE METAL LANTERN']\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+\n",
      "|                 key| value|\n",
      "+--------------------+------+\n",
      "|WHITE HANGING HEA...|536365|\n",
      "| WHITE METAL LANTERN|536365|\n",
      "+--------------------+------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(create_map(col(\"Description\"), col(\"InvoiceNo\")).alias(\"complex_map\"))\\\n",
    ".selectExpr(\"explode(complex_map)\").show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### User-Defined Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8.0"
      ]
     },
     "execution_count": 115,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "udfExampleDF = spark.range(5).toDF(\"num\")\n",
    "def power3(double_value):\n",
    "    return double_value ** 3\n",
    "power3(2.0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import udf\n",
    "power3udf = udf(power3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|power3(num)|\n",
      "+-----------+\n",
      "|          0|\n",
      "|          1|\n",
      "+-----------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import col\n",
    "udfExampleDF.select(power3udf(col(\"num\"))).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<function __main__.power3(double_value)>"
      ]
     },
     "execution_count": 120,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.types import IntegerType, DoubleType\n",
    "spark.udf.register(\"power3py\", power3, DoubleType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+\n",
      "|power3py(num)|\n",
      "+-------------+\n",
      "|         null|\n",
      "|         null|\n",
      "+-------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "udfExampleDF.selectExpr(\"power3py(num)\").show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Aggregation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[InvoiceNo: string, StockCode: string, Description: string, Quantity: int, InvoiceDate: string, UnitPrice: double, CustomerID: int, Country: string]"
      ]
     },
     "execution_count": 123,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = spark.read.format(\"csv\")\\\n",
    ".option(\"header\", \"true\")\\\n",
    ".option(\"inferSchema\", \"true\")\\\n",
    ".load(\"data/retail-data/all/*.csv\")\\\n",
    ".coalesce(5)\n",
    "df.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+--------------------+--------+--------------+---------+----------+--------------+\n",
      "|InvoiceNo|StockCode|         Description|Quantity|   InvoiceDate|UnitPrice|CustomerID|       Country|\n",
      "+---------+---------+--------------------+--------+--------------+---------+----------+--------------+\n",
      "|   536365|   85123A|WHITE HANGING HEA...|       6|12/1/2010 8:26|     2.55|     17850|United Kingdom|\n",
      "|   536365|    71053| WHITE METAL LANTERN|       6|12/1/2010 8:26|     3.39|     17850|United Kingdom|\n",
      "|   536365|   84406B|CREAM CUPID HEART...|       8|12/1/2010 8:26|     2.75|     17850|United Kingdom|\n",
      "|   536365|   84029G|KNITTED UNION FLA...|       6|12/1/2010 8:26|     3.39|     17850|United Kingdom|\n",
      "|   536365|   84029E|RED WOOLLY HOTTIE...|       6|12/1/2010 8:26|     3.39|     17850|United Kingdom|\n",
      "|   536365|    22752|SET 7 BABUSHKA NE...|       2|12/1/2010 8:26|     7.65|     17850|United Kingdom|\n",
      "|   536365|    21730|GLASS STAR FROSTE...|       6|12/1/2010 8:26|     4.25|     17850|United Kingdom|\n",
      "|   536366|    22633|HAND WARMER UNION...|       6|12/1/2010 8:28|     1.85|     17850|United Kingdom|\n",
      "|   536366|    22632|HAND WARMER RED P...|       6|12/1/2010 8:28|     1.85|     17850|United Kingdom|\n",
      "|   536367|    84879|ASSORTED COLOUR B...|      32|12/1/2010 8:34|     1.69|     13047|United Kingdom|\n",
      "|   536367|    22745|POPPY'S PLAYHOUSE...|       6|12/1/2010 8:34|      2.1|     13047|United Kingdom|\n",
      "|   536367|    22748|POPPY'S PLAYHOUSE...|       6|12/1/2010 8:34|      2.1|     13047|United Kingdom|\n",
      "|   536367|    22749|FELTCRAFT PRINCES...|       8|12/1/2010 8:34|     3.75|     13047|United Kingdom|\n",
      "|   536367|    22310|IVORY KNITTED MUG...|       6|12/1/2010 8:34|     1.65|     13047|United Kingdom|\n",
      "|   536367|    84969|BOX OF 6 ASSORTED...|       6|12/1/2010 8:34|     4.25|     13047|United Kingdom|\n",
      "|   536367|    22623|BOX OF VINTAGE JI...|       3|12/1/2010 8:34|     4.95|     13047|United Kingdom|\n",
      "|   536367|    22622|BOX OF VINTAGE AL...|       2|12/1/2010 8:34|     9.95|     13047|United Kingdom|\n",
      "|   536367|    21754|HOME BUILDING BLO...|       3|12/1/2010 8:34|     5.95|     13047|United Kingdom|\n",
      "|   536367|    21755|LOVE BUILDING BLO...|       3|12/1/2010 8:34|     5.95|     13047|United Kingdom|\n",
      "|   536367|    21777|RECIPE BOX WITH M...|       4|12/1/2010 8:34|     7.95|     13047|United Kingdom|\n",
      "+---------+---------+--------------------+--------+--------------+---------+----------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregation Functions\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|count(StockCode)|\n",
      "+----------------+\n",
      "|          541909|\n",
      "+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import count\n",
    "df.select(count(\"StockCode\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+\n",
      "|count(DISTINCT StockCode)|\n",
      "+-------------------------+\n",
      "|                     4070|\n",
      "+-------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import countDistinct\n",
    "df.select(countDistinct(\"StockCode\")).show() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+\n",
      "|approx_count_distinct(StockCode)|\n",
      "+--------------------------------+\n",
      "|                            3364|\n",
      "+--------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import approx_count_distinct\n",
    "df.select(approx_count_distinct(\"StockCode\", 0.1)).show() # 3364"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------+----------------------+\n",
      "|first(StockCode, false)|last(StockCode, false)|\n",
      "+-----------------------+----------------------+\n",
      "|                 85123A|                 22138|\n",
      "+-----------------------+----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import first, last\n",
    "df.select(first(\"StockCode\"), last(\"StockCode\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 129,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-------------+\n",
      "|min(Quantity)|max(Quantity)|\n",
      "+-------------+-------------+\n",
      "|       -80995|        80995|\n",
      "+-------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import min, max\n",
    "df.select(min(\"Quantity\"), max(\"Quantity\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+\n",
      "|sum(Quantity)|\n",
      "+-------------+\n",
      "|      5176450|\n",
      "+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import sum\n",
    "df.select(sum(\"Quantity\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------+\n",
      "|sum(DISTINCT Quantity)|\n",
      "+----------------------+\n",
      "|                 29310|\n",
      "+----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import sumDistinct\n",
    "df.select(sumDistinct(\"Quantity\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 132,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------------+----------------+----------------+\n",
      "|(total_purchases / total_transactions)|   avg_purchases|  mean_purchases|\n",
      "+--------------------------------------+----------------+----------------+\n",
      "|                      9.55224954743324|9.55224954743324|9.55224954743324|\n",
      "+--------------------------------------+----------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import sum, count, avg, expr\n",
    "df.select(\n",
    "count(\"Quantity\").alias(\"total_transactions\"),\n",
    "sum(\"Quantity\").alias(\"total_purchases\"),\n",
    "avg(\"Quantity\").alias(\"avg_purchases\"),\n",
    "expr(\"mean(Quantity)\").alias(\"mean_purchases\"))\\\n",
    ".selectExpr(\n",
    "\"total_purchases/total_transactions\",\n",
    "\"avg_purchases\",\n",
    "\"mean_purchases\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+------------------+--------------------+---------------------+\n",
      "|var_pop(Quantity)|var_samp(Quantity)|stddev_pop(Quantity)|stddev_samp(Quantity)|\n",
      "+-----------------+------------------+--------------------+---------------------+\n",
      "|47559.30364660923| 47559.39140929892|  218.08095663447835|   218.08115785023455|\n",
      "+-----------------+------------------+--------------------+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import var_pop, stddev_pop\n",
    "from pyspark.sql.functions import var_samp, stddev_samp\n",
    "df.select(var_pop(\"Quantity\"), var_samp(\"Quantity\"),\n",
    "stddev_pop(\"Quantity\"), stddev_samp(\"Quantity\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+\n",
      "|  skewness(Quantity)|kurtosis(Quantity)|\n",
      "+--------------------+------------------+\n",
      "|-0.26407557610528376|119768.05495530753|\n",
      "+--------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import skewness, kurtosis\n",
    "df.select(skewness(\"Quantity\"), kurtosis(\"Quantity\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 135,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+-------------------------------+------------------------------+\n",
      "|corr(InvoiceNo, Quantity)|covar_samp(InvoiceNo, Quantity)|covar_pop(InvoiceNo, Quantity)|\n",
      "+-------------------------+-------------------------------+------------------------------+\n",
      "|     4.912186085640497E-4|             1052.7280543915997|            1052.7260778754955|\n",
      "+-------------------------+-------------------------------+------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import corr, covar_pop, covar_samp\n",
    "df.select(corr(\"InvoiceNo\", \"Quantity\"), covar_samp(\"InvoiceNo\", \"Quantity\"),\n",
    "covar_pop(\"InvoiceNo\", \"Quantity\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 136,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------------------+\n",
      "|collect_set(Country)|collect_list(Country)|\n",
      "+--------------------+---------------------+\n",
      "|[Portugal, Italy,...| [United Kingdom, ...|\n",
      "+--------------------+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import collect_set, collect_list\n",
    "df.agg(collect_set(\"Country\"), collect_list(\"Country\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 137,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+-----+\n",
      "|InvoiceNo|CustomerId|count|\n",
      "+---------+----------+-----+\n",
      "|   536846|     14573|   76|\n",
      "|   537026|     12395|   12|\n",
      "|   537883|     14437|    5|\n",
      "|   538068|     17978|   12|\n",
      "|   538279|     14952|    7|\n",
      "|   538800|     16458|   10|\n",
      "|   538942|     17346|   12|\n",
      "|  C539947|     13854|    1|\n",
      "|   540096|     13253|   16|\n",
      "|   540530|     14755|   27|\n",
      "|   541225|     14099|   19|\n",
      "|   541978|     13551|    4|\n",
      "|   542093|     17677|   16|\n",
      "|   543188|     12567|   63|\n",
      "|   543590|     17377|   19|\n",
      "|  C543757|     13115|    1|\n",
      "|  C544318|     12989|    1|\n",
      "|   544578|     12365|    1|\n",
      "|   545165|     16339|   20|\n",
      "|   545289|     14732|   30|\n",
      "+---------+----------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"InvoiceNo\", \"CustomerId\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 138,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----+---------------+\n",
      "|InvoiceNo|quan|count(Quantity)|\n",
      "+---------+----+---------------+\n",
      "|   536596|   6|              6|\n",
      "|   536938|  14|             14|\n",
      "|   537252|   1|              1|\n",
      "|   537691|  20|             20|\n",
      "|   538041|   1|              1|\n",
      "|   538184|  26|             26|\n",
      "|   538517|  53|             53|\n",
      "|   538879|  19|             19|\n",
      "|   539275|   6|              6|\n",
      "|   539630|  12|             12|\n",
      "|   540499|  24|             24|\n",
      "|   540540|  22|             22|\n",
      "|  C540850|   1|              1|\n",
      "|   540976|  48|             48|\n",
      "|   541432|   4|              4|\n",
      "|   541518| 101|            101|\n",
      "|   541783|  35|             35|\n",
      "|   542026|   9|              9|\n",
      "|   542375|   6|              6|\n",
      "|  C542604|   8|              8|\n",
      "+---------+----+---------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"InvoiceNo\").agg(\n",
    "count(\"Quantity\").alias(\"quan\"),\n",
    "expr(\"count(Quantity)\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 139,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+------------------+--------------------+\n",
      "|InvoiceNo|     avg(Quantity)|stddev_pop(Quantity)|\n",
      "+---------+------------------+--------------------+\n",
      "|   536596|               1.5|  1.1180339887498947|\n",
      "|   536938|33.142857142857146|  20.698023172885524|\n",
      "|   537252|              31.0|                 0.0|\n",
      "|   537691|              8.15|   5.597097462078001|\n",
      "|   538041|              30.0|                 0.0|\n",
      "|   538184|12.076923076923077|   8.142590198943392|\n",
      "|   538517|3.0377358490566038|  2.3946659604837897|\n",
      "|   538879|21.157894736842106|  11.811070444356483|\n",
      "|   539275|              26.0|  12.806248474865697|\n",
      "|   539630|20.333333333333332|  10.225241100118645|\n",
      "|   540499|              3.75|  2.6653642652865788|\n",
      "|   540540|2.1363636363636362|  1.0572457590557278|\n",
      "|  C540850|              -1.0|                 0.0|\n",
      "|   540976|10.520833333333334|   6.496760677872902|\n",
      "|   541432|             12.25|  10.825317547305483|\n",
      "|   541518| 23.10891089108911|  20.550782784878713|\n",
      "|   541783|11.314285714285715|   8.467657556242811|\n",
      "|   542026| 7.666666666666667|   4.853406592853679|\n",
      "|   542375|               8.0|  3.4641016151377544|\n",
      "|  C542604|              -8.0|  15.173990905493518|\n",
      "+---------+------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"InvoiceNo\").agg(expr(\"avg(Quantity)\"),expr(\"stddev_pop(Quantity)\"))\\\n",
    ".show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Joins\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 141,
   "metadata": {},
   "outputs": [],
   "source": [
    "person = spark.createDataFrame([\n",
    "(0, \"Bill Chambers\", 0, [100]),\n",
    "(1, \"Matei Zaharia\", 1, [500, 250, 100]),\n",
    "(2, \"Michael Armbrust\", 1, [250, 100])])\\\n",
    ".toDF(\"id\", \"name\", \"graduate_program\", \"spark_status\")\n",
    "graduateProgram = spark.createDataFrame([\n",
    "(0, \"Masters\", \"School of Information\", \"UC Berkeley\"),\n",
    "(2, \"Masters\", \"EECS\", \"UC Berkeley\"),\n",
    "(1, \"Ph.D.\", \"EECS\", \"UC Berkeley\")])\\\n",
    ".toDF(\"id\", \"degree\", \"department\", \"school\")\n",
    "sparkStatus = spark.createDataFrame([\n",
    "(500, \"Vice President\"),\n",
    "(250, \"PMC Member\"),\n",
    "(100, \"Contributor\")])\\\n",
    ".toDF(\"id\", \"status\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 142,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------------+----------------+---------------+\n",
      "| id|            name|graduate_program|   spark_status|\n",
      "+---+----------------+----------------+---------------+\n",
      "|  0|   Bill Chambers|               0|          [100]|\n",
      "|  1|   Matei Zaharia|               1|[500, 250, 100]|\n",
      "|  2|Michael Armbrust|               1|     [250, 100]|\n",
      "+---+----------------+----------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "person.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 143,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+--------------------+-----------+\n",
      "| id| degree|          department|     school|\n",
      "+---+-------+--------------------+-----------+\n",
      "|  0|Masters|School of Informa...|UC Berkeley|\n",
      "|  2|Masters|                EECS|UC Berkeley|\n",
      "|  1|  Ph.D.|                EECS|UC Berkeley|\n",
      "+---+-------+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "graduateProgram.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 144,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------+\n",
      "| id|        status|\n",
      "+---+--------------+\n",
      "|500|Vice President|\n",
      "|250|    PMC Member|\n",
      "|100|   Contributor|\n",
      "+---+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sparkStatus.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 145,
   "metadata": {},
   "outputs": [],
   "source": [
    "joinExpression = person[\"graduate_program\"] == graduateProgram['id']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "metadata": {},
   "outputs": [],
   "source": [
    "wrongJoinExpression = person[\"name\"] == graduateProgram[\"school\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "| id|            name|graduate_program|   spark_status| id| degree|          department|     school|\n",
      "+---+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "|  0|   Bill Chambers|               0|          [100]|  0|Masters|School of Informa...|UC Berkeley|\n",
      "|  1|   Matei Zaharia|               1|[500, 250, 100]|  1|  Ph.D.|                EECS|UC Berkeley|\n",
      "|  2|Michael Armbrust|               1|     [250, 100]|  1|  Ph.D.|                EECS|UC Berkeley|\n",
      "+---+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinType = \"inner\"\n",
    "person.join(graduateProgram, joinExpression, joinType).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 148,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "|  id|            name|graduate_program|   spark_status| id| degree|          department|     school|\n",
      "+----+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "|   0|   Bill Chambers|               0|          [100]|  0|Masters|School of Informa...|UC Berkeley|\n",
      "|   1|   Matei Zaharia|               1|[500, 250, 100]|  1|  Ph.D.|                EECS|UC Berkeley|\n",
      "|   2|Michael Armbrust|               1|     [250, 100]|  1|  Ph.D.|                EECS|UC Berkeley|\n",
      "|null|            null|            null|           null|  2|Masters|                EECS|UC Berkeley|\n",
      "+----+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinType = \"outer\"\n",
    "person.join(graduateProgram, joinExpression, joinType).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+--------------------+-----------+----+----------------+----------------+---------------+\n",
      "| id| degree|          department|     school|  id|            name|graduate_program|   spark_status|\n",
      "+---+-------+--------------------+-----------+----+----------------+----------------+---------------+\n",
      "|  0|Masters|School of Informa...|UC Berkeley|   0|   Bill Chambers|               0|          [100]|\n",
      "|  1|  Ph.D.|                EECS|UC Berkeley|   1|   Matei Zaharia|               1|[500, 250, 100]|\n",
      "|  1|  Ph.D.|                EECS|UC Berkeley|   2|Michael Armbrust|               1|     [250, 100]|\n",
      "|  2|Masters|                EECS|UC Berkeley|null|            null|            null|           null|\n",
      "+---+-------+--------------------+-----------+----+----------------+----------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinType = \"left_outer\"\n",
    "graduateProgram.join(person, joinExpression, joinType).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 150,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "|  id|            name|graduate_program|   spark_status| id| degree|          department|     school|\n",
      "+----+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "|   0|   Bill Chambers|               0|          [100]|  0|Masters|School of Informa...|UC Berkeley|\n",
      "|   1|   Matei Zaharia|               1|[500, 250, 100]|  1|  Ph.D.|                EECS|UC Berkeley|\n",
      "|   2|Michael Armbrust|               1|     [250, 100]|  1|  Ph.D.|                EECS|UC Berkeley|\n",
      "|null|            null|            null|           null|  2|Masters|                EECS|UC Berkeley|\n",
      "+----+----------------+----------------+---------------+---+-------+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinType = \"right_outer\"\n",
    "person.join(graduateProgram, joinExpression, joinType).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+--------------------+-----------+\n",
      "| id| degree|          department|     school|\n",
      "+---+-------+--------------------+-----------+\n",
      "|  0|Masters|School of Informa...|UC Berkeley|\n",
      "|  1|  Ph.D.|                EECS|UC Berkeley|\n",
      "+---+-------+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinType = \"left_semi\"\n",
    "graduateProgram.join(person, joinExpression, joinType).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 152,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+--------------------+-----------------+\n",
      "| id| degree|          department|           school|\n",
      "+---+-------+--------------------+-----------------+\n",
      "|  0|Masters|School of Informa...|      UC Berkeley|\n",
      "|  2|Masters|                EECS|      UC Berkeley|\n",
      "|  1|  Ph.D.|                EECS|      UC Berkeley|\n",
      "|  0|Masters|      Duplicated Row|Duplicated School|\n",
      "+---+-------+--------------------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "gradProgram2 = graduateProgram.union(spark.createDataFrame([\n",
    "(0, \"Masters\", \"Duplicated Row\", \"Duplicated School\")]))\n",
    "gradProgram2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 153,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+----------+-----------+\n",
      "| id| degree|department|     school|\n",
      "+---+-------+----------+-----------+\n",
      "|  2|Masters|      EECS|UC Berkeley|\n",
      "+---+-------+----------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinType = \"left_anti\"\n",
    "graduateProgram.join(person, joinExpression, joinType).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 154,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+--------------------+-----------+---+----------------+----------------+---------------+\n",
      "| id| degree|          department|     school| id|            name|graduate_program|   spark_status|\n",
      "+---+-------+--------------------+-----------+---+----------------+----------------+---------------+\n",
      "|  0|Masters|School of Informa...|UC Berkeley|  0|   Bill Chambers|               0|          [100]|\n",
      "|  1|  Ph.D.|                EECS|UC Berkeley|  1|   Matei Zaharia|               1|[500, 250, 100]|\n",
      "|  1|  Ph.D.|                EECS|UC Berkeley|  2|Michael Armbrust|               1|     [250, 100]|\n",
      "+---+-------+--------------------+-----------+---+----------------+----------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinType = \"cross\"\n",
    "graduateProgram.join(person, joinExpression, joinType).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#  Spark SQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To use SQL queries directly with the dataframe, you will need to register it to a temporary view:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 156,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.read.json(\"data/2015-summary.json\")\\\n",
    ".createOrReplaceTempView(\"some_sql_view\") # DF => SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 157,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "12"
      ]
     },
     "execution_count": 157,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT DEST_COUNTRY_NAME, sum(count)\n",
    "FROM some_sql_view GROUP BY DEST_COUNTRY_NAME\n",
    "\"\"\")\\\n",
    ".where(\"DEST_COUNTRY_NAME like 'S%'\").where(\"`sum(count)` > 10\")\\\n",
    ".count() # SQL => DF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.end()"
   ]
  }
 ],
 "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.8.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}