Colab and PySpark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "iox_ufgbqDXa"
   },
   "source": [
    "<h1><center>Introduction to Google Colab and PySpark</center></h1>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "1qV6Grv7qIa9"
   },
   "source": [
    "## Table Of Contents:\n",
    "<ol>\n",
    "<li><a href=\"#objective\">Objective</a></li>\n",
    "<li><a href=\"#prerequisite\">Prerequisite</a></li>\n",
    "<li><a href=\"#notes-from-the-author\">Notes from the Author</a></li>\n",
    "<li><a href=\"#big-data-pyspark-and-colaboratory\">Big data, PySpark and Colaboratory</a>\n",
    "    <ol>\n",
    "        <li><a href=\"#big-data\">Big data</a></li>\n",
    "        <li><a href=\"#pyspark\">PySpark</a></li>\n",
    "        <li><a href=\"#colaboratory\">Colaboratory</a></li>\n",
    "    </ol>\n",
    "</li>\n",
    "<li><a href=\"#jupyter-notebook-basics\">Jupyter Notebook Basics</a>\n",
    "    <ol>\n",
    "        <li><a href=\"#code-cells\">Code cells</a></li>\n",
    "        <li><a href=\"#text-cells\">Text cells</a></li>\n",
    "        <li><a href=\"#access-to-the-shell\">Access to the shell</a></li>\n",
    "        <li><a href=\"#installing-spark\">Installing Spark</a></li>\n",
    "    </ol>\n",
    "</li>\n",
    "<li><a href=\"#exploring-the-dataset\">Exploring the Dataset</a>\n",
    "    <ol>\n",
    "        <li><a href=\"#loading-the-dataset\">Loading the Dataset</a></li>\n",
    "        <li><a href=\"#viewing-the-dataframe\">Viewing the Dataframe</a></li>\n",
    "        <li><a href=\"#viewing-dataframe-columns\">Viewing Dataframe Columns</a></li>\n",
    "        <li><a href=\"#dataframe-schema\">Dataframe Schema</a>\n",
    "          <ul>\n",
    "            <li><a href=\"#implicit-schema-inference\">Inferring Schema Implicitly</a></li>\n",
    "            <li><a href=\"#explicit-schema-inference\">Defining Schema Explicitly</a></li>\n",
    "          </ul>\n",
    "        </li>\n",
    "    </ol>\n",
    "</li>\n",
    "<li><a href=\"#dataframe-operations-on-columns\">DataFrame Operations on Columns</a>\n",
    "    <ol>\n",
    "        <li><a href=\"#selecting-columns\">Selecting Columns</a></li>\n",
    "        <li><a href=\"#selecting-multiple-columns\">Selecting Multiple Columns</a></li>\n",
    "        <li><a href=\"#adding-new-columns\">Adding New Columns</a></li>\n",
    "        <li><a href=\"#renaming-columns\">Renaming Columns</a>\n",
    "        <li><a href=\"#grouping-by-columns\">Grouping By Columns</a>\n",
    "        <li><a href=\"#removing-columns\">Removing Columns</a>\n",
    "    </ol>\n",
    "</li>\n",
    "<li><a href=\"#dataframe-operations-on-rows\">DataFrame Operations on Rows</a>\n",
    "    <ol>\n",
    "        <li><a href=\"#filtering-rows\">Filtering Rows</a></li>\n",
    "        <li><a href=\"#get-distinct-rows\">Get Distinct Rows</a></li>\n",
    "        <li><a href=\"#sorting-rows\">Sorting Rows</a></li>\n",
    "        <li><a href=\"#union-dataframes\">Union Dataframes</a>\n",
    "    </ol>\n",
    "</li>\n",
    "<li><a href=\"#common-data-manipulation-functions\">Common Data Manipulation Functions</a>\n",
    "    <ol>\n",
    "        <li><a href=\"#string-functions\">String Functions</a></li>\n",
    "        <li><a href=\"#numeric-functions\">Numeric Functions</a></li>\n",
    "        <li><a href=\"#operations-on-date\">Operations on Date</a></li>\n",
    "    </ol>\n",
    "</li>\n",
    "<li><a href=\"#joins-in-pyspark\">Joins in PySpark</a></li>\n",
    "<li><a href=\"#spark-sql\">Spark SQL</a></li>\n",
    "<li><a href=\"#rdd\">RDD</a></li>\n",
    "<li><a href=\"#user-defined-functions-udf\">User-Defined Functions (UDF)</a></li>\n",
    "<li><a href=\"#common-questions\">Common Questions</a>\n",
    "    <ol>\n",
    "        <li><a href=\"#recommended-ide\">Recommended IDE</a></li>\n",
    "        <li><a href=\"#submitting-a-spark-job\">Submitting a Spark Job</a></li>\n",
    "        <li><a href=\"#creating-dataframes\">Creating Dataframes</a></li>\n",
    "        <li><a href=\"#drop-duplicates\">Drop Duplicates</a></li>\n",
    "        <li><a href=\"#fine-tuning-a-pyspark-job\">Fine Tuning a PySpark Job</a>\n",
    "          <ul>\n",
    "            <li><a href=\"#emr-sizing\">EMR Sizing</a></li>\n",
    "            <li><a href=\"#spark-configurations\">Spark Configurations</a></li>\n",
    "            <li><a href=\"#job-tuning\">Job Tuning</a>\n",
    "            <li><a href=\"#best-practices\">Best Practices</a>\n",
    "          </ul>\n",
    "        </li>\n",
    "    </ol>\n",
    "</li>\n",
    "</ol>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "FFnYZltvqLgt"
   },
   "source": [
    "<a id='objective'></a>\n",
    "## Objective\n",
    "The objective of this notebook is to:\n",
    "><li>Give a proper understanding about the different PySpark functions available. </li>\n",
    "><li>A short introduction to Google Colab, as that is the platform on which this notebook is written on. </li>\n",
    "\n",
    "Once you complete this notebook, you should be able to write pyspark programs in an efficent way. The ideal way to use this is by going through the examples given and then trying them on Colab. At the end there are a few hands on questions which you can use to evaluate yourself."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "YR1CO3FTqO5h"
   },
   "source": [
    "<a id='prerequisite'></a>\n",
    "## Prerequisite\n",
    "><li>Although some theory about pyspark and big data will be given in this notebook, I recommend everyone to read more about it and have a deeper understanding on how the functions get executed and the relevance of big data in the current scenario.\n",
    "><li>A good understanding on python will be an added bonus."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "bGbIBPLHqVXc"
   },
   "source": [
    "<a id='notes-from-the-author'></a>\n",
    "## Notes from the Author\n",
    "\n",
    "This tutorial was made using Google Colab so the code you see here is meant to run on a colab notebook. <br>\n",
    "It goes through basic [PySpark Functions](https://spark.apache.org/docs/latest/api/python/index.html) and a short introduction on how to use [Colab](https://colab.research.google.com/notebooks/basic_features_overview.ipynb). <br>\n",
    "If you want to view my colab notebook for this particular tutorial, you can view it [here](https://colab.research.google.com/drive/1G894WS7ltIUTusWWmsCnF_zQhQqZCDOc). The viewing experience and readability is much better there. <br>\n",
    "If you want to try out things with this notebook as a base, feel free to download it from my repo [here](https://github.com/jacobceles/knowledge-repo/blob/master/pyspark/Colab%20and%20PySpark.ipynb) and then use it with jupyter notebook."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "K_q3Yzc9qYc0"
   },
   "source": [
    "<a id='big-data-pyspark-and-colaboratory'></a>\n",
    "## Big data, PySpark and Colaboratory"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "5gs9JXCWqb9s"
   },
   "source": [
    "<a id='big-data'></a>\n",
    "### Big data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "q3UkLT6yqebl"
   },
   "source": [
    "Big data usually means data of such huge volume that normal data storage solutions cannot efficently store and process it. In this era, data is being generated at an absurd rate. Data is collected for each movement a person makes. The bulk of big data comes from three primary sources: \n",
    "<ol>\n",
    "   <li>Social data</li> \n",
    "   <li>Machine data</li> \n",
    "   <li>Transactional data</li>\n",
    "</ol>\n",
    "\n",
    "Some common examples for the sources of such data include internet searches, facebook posts, doorbell cams, smartwatches, online shopping history etc. Every action creates data, it is just a matter of of there is a way to collect them or not.  But what's interesting is that out of all this data collected, not even 5% of it is being used fully. There is a huge demand for big data professionals in the industry. Even though the number of graduates with a specialization in big data are rising, the problem is that they don't have the practical knowledge about big data scenarios, which leads to bad architecutres and inefficent methods of processing data.\n",
    "\n",
    ">If you are interested to know more about the landscape and technologies involved, here is [an article](https://hostingtribunal.com/blog/big-data-stats/) which I found really interesting!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "NhM3wLG2qhlN"
   },
   "source": [
    "<a id='pyspark'></a>\n",
    "### PySpark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "VBfC_kvjqjPj"
   },
   "source": [
    "If you are working in the field of big data, you must have definelty heard of spark. If you look at the [Apache Spark](https://spark.apache.org/) website, you will see that it is said to be a `Lightning-fast unified analytics engine`. PySpark is a flavour of Spark used for processing and analysing massive volumes of data. If you are familiar with python and have tried it for huge datasets, you should know that the execution time can get ridiculous. Enter PySpark!\n",
    "\n",
    "Imagine your data resides in a distributed manner at different places. If you try brining your data to one point and executing your code there, not only would that be inefficent, but also cause memory issues. Now let's say your code goes to the data rather than the data coming to where your code. This will help avoid unneccesary data movement which will thereby decrease the running time. \n",
    "\n",
    "PySpark is the Python API of Spark; which means it can do almost all the things python can. Machine learning(ML) pipelines, exploratory data analysis (at scale), ETLs for data platform, and much more! And all of them in a distributed manner. One of the best parts of pyspark is that if you are already familiar with python, it's really easy to learn.\n",
    "\n",
    "Apart from PySpark, there is another language called Scala used for big data processing. Scala is frequently over 10 times faster than *Python*, as it is native for Hadoop as its based on JVM. But PySpark is getting adopted at a fast rate because of the ease of use, easier learning curve and ML capabilities.\n",
    "\n",
    "I will briefly explain how a PySpark job works, but I strongly recommend you read more about the [architecture](https://data-flair.training/blogs/how-apache-spark-works/) and how everything works. Now, before I get into it, let me talk about some <u>basic jargons</u> first:\n",
    "\n",
    "<b>Cluster</b> is a set of loosely or tightly connected computers that work together so that they can be viewed as a single system.\n",
    "\n",
    "<b>Hadoop</b> is an open source, scalable, and fault tolerant framework written in Java. It efficiently processes large volumes of data on a cluster of commodity hardware. Hadoop is not only a storage system but is a platform for large data storage as well as processing.\n",
    "\n",
    "<b>HDFS</b> (Hadoop distributed file system). It is one of the world's most reliable storage system. HDFS is a Filesystem of Hadoop designed for storing very large files running on a cluster of commodity hardware.\n",
    "\n",
    "<b>MapReduce</b> is a data Processing framework, which has 2 phases - Mapper and Reducer. The map procedure performs filtering and sorting, and the reduce method performs a summary operation. It usually runs on a hadoop cluster.\n",
    "\n",
    "<b>Transformation</b> refers to the operations applied on a dataset to create a new dataset. Filter, groupBy and map are the examples of transformations.\n",
    "\n",
    "<b>Actions</b> Actions refer to an operation which instructs Spark to perform computation and send the result back to driver. This is an example of action.\n",
    "\n",
    "Alright! Now that that's out of the way, let me explain how a spark job runs. In simple terma, each time you submit a pyspark job, the code gets internally converted into a MapReduce program and gets executed in the Java virtual machine. Now one of the thoughts that might be popping in your mind will probably be: <br>`So the code gets converted into a MapReduce program. Wouldn't that mean MapReduce is faster than pySpark?`<br> Well, the answer is a big NO. This is what makes spark jobs special. Spark is capable of handling a massive amount of data at a time, in it's distributed environment. It does this through <u>in-memory processing</u>, which is what makes it almost 100 times faster than Hadoop. Another factor which amkes it fast is <u>Lazy Evaluation</u>. Spark delays its evaluation as much as it can. Each time you  submit a job, spark creates an action plan for how to execute the code, and then does nothing. Finally, when you ask for the result(i.e, calls an action), it executes the plan, which is basically all the transofrmations you have mentioned in your code. That's basically the gist of it. \n",
    "\n",
    "Now lastly, I want to talk about on more thing. Spark mainly consists of 4 modules:\n",
    "\n",
    "<ol>\n",
    "    <li>Spark SQL - helps to write  spark programs using SQL like queries.</li>\n",
    "    <li>Spark Streaming - is an extension of the core Spark API that enables scalable, high-throughput, fault-tolerant stream processing of live data streams. used heavily in processing of social media data.</li>\n",
    "    <li>Spark MLLib - is the machine learning component of SPark. It helps train ML models on massive datasets with very high efficeny. </li>\n",
    "    <li>Spark GraphX - is the visualization component of Spark. It enables users to view data both as graphs and as collections without data movement or duplication.</li>\n",
    "</ol>\n",
    "\n",
    "Hopefully this image gives a better idea of what I am talking about:\n",
    "<img alt=\"Spark Modules\" src=\"https://2s7gjr373w3x22jf92z99mgm5w-wpengine.netdna-ssl.com/wp-content/uploads/2015/11/spark-streaming-datanami.png\" />\n",
    "<center><font color='#666956'>Source: Datanami</font><center>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "1NJMWs4NqnlF"
   },
   "source": [
    "<a id='colaboratory'></a>\n",
    "### Colaboratory"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ynKFk6b7qoWr"
   },
   "source": [
    "In the words of Google: <br>\n",
    "`Colaboratory, or “Colab” for short, is a product from Google Research. Colab allows anybody to write and execute arbitrary python code through the browser, and is especially well suited to machine learning, data analysis and education. More technically, Colab is a hosted Jupyter notebook service that requires no setup to use, while providing free access to computing resources including GPUs.`\n",
    "\n",
    "The reason why I used colab is because of its shareability and free GPU and TPU. Yeah you read that right, FREE GPU AND TPU! For using TPU, your program needs to be optimized for the same. Additionally, it helps use different Google services conveniently. It saves to Google Drive and all the services are very closely related. I recommend you go through the offical [overview documentation](https://colab.research.google.com/notebooks/basic_features_overview.ipynb) if you want to know more about it.\n",
    "If you have more questions about colab, please [refer this link](https://research.google.com/colaboratory/faq.html).\n",
    "\n",
    ">While using a colab notebook, you will need an active internet connection to keep a session alive. If you lose the connection you will have to download the datasets again."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "_N5-lspH_N8B"
   },
   "source": [
    "<a id='jupyter-notebook-basics'></a>\n",
    "## Jupyter notebook basics"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "6Ul54hAYyHyd"
   },
   "source": [
    "<a id='code-cells'></a>\n",
    "### Code cells"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "j38beRUTCI5c",
    "outputId": "5a5d549b-afa7-4742-84ce-f9bec67dd9a9"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6"
      ]
     },
     "execution_count": 1,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "2*3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "id": "_Jewe_e9CIYa"
   },
   "outputs": [],
   "source": [
    "from collections import Counter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "g8Y7w6_CCIIT",
    "outputId": "ee375530-ba9b-4b26-ac91-4c3aedb8a11e"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "This is a tutorial!\n"
     ]
    }
   ],
   "source": [
    "print(\"This is a tutorial!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "VOqLNkRKyUIS"
   },
   "source": [
    "<a id='text-cells'></a>\n",
    "### Text cells"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "IfbaUe-oq7DK"
   },
   "source": [
    "Hello world!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "X6zdrH15_CCW"
   },
   "source": [
    "<a id='access-to-the-shell'></a>\n",
    "### Access to the shell"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "zdO9sjSdEVnr",
    "outputId": "727c5cf0-3286-462b-a259-a550011b8460"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[0m\u001b[01;34msample_data\u001b[0m/\n"
     ]
    }
   ],
   "source": [
    "ls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 35
    },
    "id": "QF9e3lDDEX3I",
    "outputId": "7ddef68f-aef1-4f46-f20a-d28ead31be97"
   },
   "outputs": [
    {
     "data": {
      "application/vnd.google.colaboratory.intrinsic+json": {
       "type": "string"
      },
      "text/plain": [
       "'/content'"
      ]
     },
     "execution_count": 5,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pwd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Dd6t0uFzuR4X"
   },
   "source": [
    "<a id='installing-spark'></a>\n",
    "### Installing Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "6apGVff5h4ca"
   },
   "source": [
    "Install Dependencies:\n",
    "\n",
    "\n",
    "1.   Java 8\n",
    "2.   Apache Spark with hadoop and\n",
    "3.   Findspark (used to locate the spark in the system)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "id": "tt7ZS1_wGgjn"
   },
   "outputs": [],
   "source": [
    "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n",
    "!wget -q http://archive.apache.org/dist/spark/spark-3.1.1/spark-3.1.1-bin-hadoop3.2.tgz\n",
    "!tar xf spark-3.1.1-bin-hadoop3.2.tgz\n",
    "!pip install -q findspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "C3x0ZRLxjMVr"
   },
   "source": [
    "Set Environment Variables:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "id": "sdOOq4twHN1K"
   },
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
    "os.environ[\"SPARK_HOME\"] = \"/content/spark-3.1.1-bin-hadoop3.2\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "3ACYMwhgHTYz",
    "outputId": "9346c054-ab0c-4559-b4b8-5867f1c8e071"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sample_data  spark-3.1.1-bin-hadoop3.2\tspark-3.1.1-bin-hadoop3.2.tgz\n"
     ]
    }
   ],
   "source": [
    "!ls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 217
    },
    "id": "KR1zLBk1998Z",
    "outputId": "79e2264b-0fd0-4238-c66c-7efea69888f9"
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://cf07049a4469:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.1.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>pyspark-shell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7fcf93fb90d0>"
      ]
     },
     "execution_count": 9,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.master(\"local[*]\").getOrCreate()\n",
    "spark.conf.set(\"spark.sql.repl.eagerEval.enabled\", True) # Property used to format output tables better\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "hmIqq6xPK7m7"
   },
   "source": [
    "<a id='exploring-the-dataset'></a>\n",
    "## Exploring the Dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "VZwsr57lwPgq"
   },
   "source": [
    "<a id='loading-the-dataset'></a>\n",
    "### Loading the Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "hQ3zmGACLKlN",
    "outputId": "1c07ff65-0536-4dac-cf14-dc910c4ebee3"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2021-03-30 18:26:09--  https://jacobceles.github.io/knowledge_repo/colab_and_pyspark/cars.csv\n",
      "Resolving jacobceles.github.io (jacobceles.github.io)... 185.199.108.153, 185.199.109.153, 185.199.110.153, ...\n",
      "Connecting to jacobceles.github.io (jacobceles.github.io)|185.199.108.153|:443... connected.\n",
      "HTTP request sent, awaiting response... 301 Moved Permanently\n",
      "Location: https://jacobcelestine.com/knowledge_repo/colab_and_pyspark/cars.csv [following]\n",
      "--2021-03-30 18:26:10--  https://jacobcelestine.com/knowledge_repo/colab_and_pyspark/cars.csv\n",
      "Resolving jacobcelestine.com (jacobcelestine.com)... 185.199.108.153, 185.199.109.153, 185.199.110.153, ...\n",
      "Connecting to jacobcelestine.com (jacobcelestine.com)|185.199.108.153|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 22608 (22K) [text/csv]\n",
      "Saving to: ‘cars.csv’\n",
      "\n",
      "cars.csv            100%[===================>]  22.08K  --.-KB/s    in 0.003s  \n",
      "\n",
      "2021-03-30 18:26:10 (7.41 MB/s) - ‘cars.csv’ saved [22608/22608]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Downloading and preprocessing Cars Data downloaded origianlly from https://perso.telecom-paristech.fr/eagan/class/igr204/datasets\n",
    "!wget https://jacobceles.github.io/knowledge_repo/colab_and_pyspark/cars.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Wpq2jYvIMOJy",
    "outputId": "a85b230e-13e6-4d29-8521-de5c99a689b9"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "cars.csv  sample_data  spark-3.1.1-bin-hadoop3.2  spark-3.1.1-bin-hadoop3.2.tgz\n"
     ]
    }
   ],
   "source": [
    "!ls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "hz6ALr5mMqZt",
    "outputId": "5a09439c-a96c-4ab5-d236-7890d65d791f"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "|                 Car| MPG|Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|\n",
      "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "|Chevrolet Chevell...|18.0|        8|       307.0|     130.0| 3504.|        12.0|   70|    US|\n",
      "|   Buick Skylark 320|15.0|        8|       350.0|     165.0| 3693.|        11.5|   70|    US|\n",
      "|  Plymouth Satellite|18.0|        8|       318.0|     150.0| 3436.|        11.0|   70|    US|\n",
      "|       AMC Rebel SST|16.0|        8|       304.0|     150.0| 3433.|        12.0|   70|    US|\n",
      "|         Ford Torino|17.0|        8|       302.0|     140.0| 3449.|        10.5|   70|    US|\n",
      "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Load data from csv to a dataframe. \n",
    "# header=True means the first row is a header \n",
    "# sep=';' means the column are seperated using ''\n",
    "df = spark.read.csv('cars.csv', header=True, sep=\";\")\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "N0lCS2LNwnoy"
   },
   "source": [
    "The above command loads our data from into a dataframe (DF). A dataframe is a 2-dimensional labeled data structure with columns of potentially different types."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "_QwZtWxZRCBn"
   },
   "source": [
    "<a id='viewing-the-dataframe'></a>\n",
    "### Viewing the Dataframe"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "50LZ3S8_PMg_"
   },
   "source": [
    "There are a couple of ways to view your dataframe(DF) in PySpark:\n",
    "\n",
    "1.   `df.take(5)` will return a list of five Row objects. \n",
    "2.   `df.collect()` will get all of the data from the entire DataFrame. Be really careful when using it, because if you have a large data set, you can easily crash the driver node. \n",
    "3.   `df.show()` is the most commonly used method to view a dataframe. There are a few parameters we can pass to this method, like the number of rows and truncaiton. For example, `df.show(5, False)` or ` df.show(5, truncate=False)` will show the entire data wihtout any truncation.\n",
    "4.   `df.limit(5)` will **return a new DataFrame** by taking the first n rows. As spark is distributed in nature, there is no guarantee that `df.limit()` will give you the same results each time.\n",
    "\n",
    "Let us see some of them in action below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "I1qqkqcfxM0v",
    "outputId": "10a5e152-8c75-4624-acfd-c86048d5de4f"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "|Car                      |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "|Chevrolet Chevelle Malibu|18.0|8        |307.0       |130.0     |3504. |12.0        |70   |US    |\n",
      "|Buick Skylark 320        |15.0|8        |350.0       |165.0     |3693. |11.5        |70   |US    |\n",
      "|Plymouth Satellite       |18.0|8        |318.0       |150.0     |3436. |11.0        |70   |US    |\n",
      "|AMC Rebel SST            |16.0|8        |304.0       |150.0     |3433. |12.0        |70   |US    |\n",
      "|Ford Torino              |17.0|8        |302.0       |140.0     |3449. |10.5        |70   |US    |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(5, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 154
    },
    "id": "R9zwzswIxXF9",
    "outputId": "033ab9bd-4064-4067-ec32-abe5dac4cc0a"
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>Car</th><th>MPG</th><th>Cylinders</th><th>Displacement</th><th>Horsepower</th><th>Weight</th><th>Acceleration</th><th>Model</th><th>Origin</th></tr>\n",
       "<tr><td>Chevrolet Chevell...</td><td>18.0</td><td>8</td><td>307.0</td><td>130.0</td><td>3504.</td><td>12.0</td><td>70</td><td>US</td></tr>\n",
       "<tr><td>Buick Skylark 320</td><td>15.0</td><td>8</td><td>350.0</td><td>165.0</td><td>3693.</td><td>11.5</td><td>70</td><td>US</td></tr>\n",
       "<tr><td>Plymouth Satellite</td><td>18.0</td><td>8</td><td>318.0</td><td>150.0</td><td>3436.</td><td>11.0</td><td>70</td><td>US</td></tr>\n",
       "<tr><td>AMC Rebel SST</td><td>16.0</td><td>8</td><td>304.0</td><td>150.0</td><td>3433.</td><td>12.0</td><td>70</td><td>US</td></tr>\n",
       "<tr><td>Ford Torino</td><td>17.0</td><td>8</td><td>302.0</td><td>140.0</td><td>3449.</td><td>10.5</td><td>70</td><td>US</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
       "|                 Car| MPG|Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|\n",
       "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
       "|Chevrolet Chevell...|18.0|        8|       307.0|     130.0| 3504.|        12.0|   70|    US|\n",
       "|   Buick Skylark 320|15.0|        8|       350.0|     165.0| 3693.|        11.5|   70|    US|\n",
       "|  Plymouth Satellite|18.0|        8|       318.0|     150.0| 3436.|        11.0|   70|    US|\n",
       "|       AMC Rebel SST|16.0|        8|       304.0|     150.0| 3433.|        12.0|   70|    US|\n",
       "|         Ford Torino|17.0|        8|       302.0|     140.0| 3449.|        10.5|   70|    US|\n",
       "+--------------------+----+---------+------------+----------+------+------------+-----+------+"
      ]
     },
     "execution_count": 14,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "eUazdCEmu_sp"
   },
   "source": [
    "<a id='viewing-dataframe-columns'></a>\n",
    "### Viewing Dataframe Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "9o7jsazcu-13",
    "outputId": "a2ce058a-2266-4333-93f3-1b071904cc0f"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Car',\n",
       " 'MPG',\n",
       " 'Cylinders',\n",
       " 'Displacement',\n",
       " 'Horsepower',\n",
       " 'Weight',\n",
       " 'Acceleration',\n",
       " 'Model',\n",
       " 'Origin']"
      ]
     },
     "execution_count": 15,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "3lfS2DhHuhPl"
   },
   "source": [
    "<a id='dataframe-schema'></a>\n",
    "### Dataframe Schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "-xX7hRoW_cXY"
   },
   "source": [
    "There are two methods commonly used to view the data types of a dataframe:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "w6qwTjGsNxrw",
    "outputId": "60faa488-aa99-4d84-d644-7e6e406e389a"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Car', 'string'),\n",
       " ('MPG', 'string'),\n",
       " ('Cylinders', 'string'),\n",
       " ('Displacement', 'string'),\n",
       " ('Horsepower', 'string'),\n",
       " ('Weight', 'string'),\n",
       " ('Acceleration', 'string'),\n",
       " ('Model', 'string'),\n",
       " ('Origin', 'string')]"
      ]
     },
     "execution_count": 16,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "CCGTFlCWRPw4",
    "outputId": "17bcb250-22c4-4d6a-f2a7-b67e99f080e7"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Car: string (nullable = true)\n",
      " |-- MPG: string (nullable = true)\n",
      " |-- Cylinders: string (nullable = true)\n",
      " |-- Displacement: string (nullable = true)\n",
      " |-- Horsepower: string (nullable = true)\n",
      " |-- Weight: string (nullable = true)\n",
      " |-- Acceleration: string (nullable = true)\n",
      " |-- Model: string (nullable = true)\n",
      " |-- Origin: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "RXx5ATpZ9oor"
   },
   "source": [
    "<a id='implicit-schema-inference'></a>\n",
    "#### Inferring Schema Implicitly"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "7TeflTUp8l29"
   },
   "source": [
    "We can use the parameter `inferschema=true` to infer the input schema automatically while loading the data. An example is shown below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Qym5MjCi894N",
    "outputId": "f92d912b-621b-4381-cd6c-2eafc89c72f7"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Car: string (nullable = true)\n",
      " |-- MPG: double (nullable = true)\n",
      " |-- Cylinders: integer (nullable = true)\n",
      " |-- Displacement: double (nullable = true)\n",
      " |-- Horsepower: double (nullable = true)\n",
      " |-- Weight: decimal(4,0) (nullable = true)\n",
      " |-- Acceleration: double (nullable = true)\n",
      " |-- Model: integer (nullable = true)\n",
      " |-- Origin: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.read.csv('cars.csv', header=True, sep=\";\", inferSchema=True)\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "G6jTedYd-Dhb"
   },
   "source": [
    "As you can see, the datatype has been infered automatically spark with even the correct precison for decimal type. A problem that might arise here is that sometimes, when you have to read multiple files with different schemas in different files, there might be an issue with implicit inferring leading to null values in some columns. Therefore, let us also see how to define schemas explicitly."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "yTVjYqeRuxWn"
   },
   "source": [
    "<a id='explicit-schema-inference'></a>\n",
    "#### Defining Schema Explicitly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "xpsaQ4JMRUiS",
    "outputId": "cbfbeed2-7aa1-43de-c8a6-5d505c164062"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Car',\n",
       " 'MPG',\n",
       " 'Cylinders',\n",
       " 'Displacement',\n",
       " 'Horsepower',\n",
       " 'Weight',\n",
       " 'Acceleration',\n",
       " 'Model',\n",
       " 'Origin']"
      ]
     },
     "execution_count": 19,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.types import *\n",
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "id": "ik62VX34SlFh"
   },
   "outputs": [],
   "source": [
    "# Creating a list of the schema in the format column_name, data_type\n",
    "labels = [\n",
    "     ('Car',StringType()),\n",
    "     ('MPG',DoubleType()),\n",
    "     ('Cylinders',IntegerType()),\n",
    "     ('Displacement',DoubleType()),\n",
    "     ('Horsepower',DoubleType()),\n",
    "     ('Weight',DoubleType()),\n",
    "     ('Acceleration',DoubleType()),\n",
    "     ('Model',IntegerType()),\n",
    "     ('Origin',StringType())\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "T-Fp5y_oU9SF",
    "outputId": "a0a84e9b-3cca-431f-ba19-d3c6c183c797"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType(List(StructField(Car,StringType,true),StructField(MPG,DoubleType,true),StructField(Cylinders,IntegerType,true),StructField(Displacement,DoubleType,true),StructField(Horsepower,DoubleType,true),StructField(Weight,DoubleType,true),StructField(Acceleration,DoubleType,true),StructField(Model,IntegerType,true),StructField(Origin,StringType,true)))"
      ]
     },
     "execution_count": 21,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Creating the schema that will be passed when reading the csv\n",
    "schema = StructType([StructField (x[0], x[1], True) for x in labels])\n",
    "schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "sgC7gtL5VTls",
    "outputId": "1a84e4b5-0c32-4d8f-f399-602a09156346"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Car: string (nullable = true)\n",
      " |-- MPG: double (nullable = true)\n",
      " |-- Cylinders: integer (nullable = true)\n",
      " |-- Displacement: double (nullable = true)\n",
      " |-- Horsepower: double (nullable = true)\n",
      " |-- Weight: double (nullable = true)\n",
      " |-- Acceleration: double (nullable = true)\n",
      " |-- Model: integer (nullable = true)\n",
      " |-- Origin: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.read.csv('cars.csv', header=True, sep=\";\", schema=schema)\n",
    "df.printSchema()\n",
    "# The schema comes as we gave!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Dn2EAhesVmx0",
    "outputId": "dc6c0aed-1731-4677-9311-708de99f3561"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "|Car                             |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|\n",
      "+--------------------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "|Chevrolet Chevelle Malibu       |18.0|8        |307.0       |130.0     |3504.0|12.0        |70   |US    |\n",
      "|Buick Skylark 320               |15.0|8        |350.0       |165.0     |3693.0|11.5        |70   |US    |\n",
      "|Plymouth Satellite              |18.0|8        |318.0       |150.0     |3436.0|11.0        |70   |US    |\n",
      "|AMC Rebel SST                   |16.0|8        |304.0       |150.0     |3433.0|12.0        |70   |US    |\n",
      "|Ford Torino                     |17.0|8        |302.0       |140.0     |3449.0|10.5        |70   |US    |\n",
      "|Ford Galaxie 500                |15.0|8        |429.0       |198.0     |4341.0|10.0        |70   |US    |\n",
      "|Chevrolet Impala                |14.0|8        |454.0       |220.0     |4354.0|9.0         |70   |US    |\n",
      "|Plymouth Fury iii               |14.0|8        |440.0       |215.0     |4312.0|8.5         |70   |US    |\n",
      "|Pontiac Catalina                |14.0|8        |455.0       |225.0     |4425.0|10.0        |70   |US    |\n",
      "|AMC Ambassador DPL              |15.0|8        |390.0       |190.0     |3850.0|8.5         |70   |US    |\n",
      "|Citroen DS-21 Pallas            |0.0 |4        |133.0       |115.0     |3090.0|17.5        |70   |Europe|\n",
      "|Chevrolet Chevelle Concours (sw)|0.0 |8        |350.0       |165.0     |4142.0|11.5        |70   |US    |\n",
      "|Ford Torino (sw)                |0.0 |8        |351.0       |153.0     |4034.0|11.0        |70   |US    |\n",
      "|Plymouth Satellite (sw)         |0.0 |8        |383.0       |175.0     |4166.0|10.5        |70   |US    |\n",
      "|AMC Rebel SST (sw)              |0.0 |8        |360.0       |175.0     |3850.0|11.0        |70   |US    |\n",
      "|Dodge Challenger SE             |15.0|8        |383.0       |170.0     |3563.0|10.0        |70   |US    |\n",
      "|Plymouth 'Cuda 340              |14.0|8        |340.0       |160.0     |3609.0|8.0         |70   |US    |\n",
      "|Ford Mustang Boss 302           |0.0 |8        |302.0       |140.0     |3353.0|8.0         |70   |US    |\n",
      "|Chevrolet Monte Carlo           |15.0|8        |400.0       |150.0     |3761.0|9.5         |70   |US    |\n",
      "|Buick Estate Wagon (sw)         |14.0|8        |455.0       |225.0     |3086.0|10.0        |70   |US    |\n",
      "+--------------------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "MDCO3TEe95OY"
   },
   "source": [
    "As we can see here, the data has been successully loaded with the specified datatypes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "rsD48rckdHPe"
   },
   "source": [
    "<a id='dataframe-operations-on-columns'></a>\n",
    "## DataFrame Operations on Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "cMlxdWfSY8ks"
   },
   "source": [
    "We will go over the following in this section:\n",
    "\n",
    "1.   Selecting Columns\n",
    "2.   Selecting Multiple Columns\n",
    "3.   Adding New Columns\n",
    "4.   Renaming Columns\n",
    "5.   Grouping By Columns\n",
    "6.   Removing Columns\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ikGR5pDICTu7"
   },
   "source": [
    "<a id='selecting-columns'></a>\n",
    "### Selecting Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "-VMwIwi2rj_o"
   },
   "source": [
    "There are multiple ways to do a select in PySpark. You can find how they differ and how each below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "ge9-_ygideWk",
    "outputId": "37bd7e10-1169-4243-fb5e-798cd980c423"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Column<'Car'>\n",
      "********************\n",
      "+--------------------------------+\n",
      "|Car                             |\n",
      "+--------------------------------+\n",
      "|Chevrolet Chevelle Malibu       |\n",
      "|Buick Skylark 320               |\n",
      "|Plymouth Satellite              |\n",
      "|AMC Rebel SST                   |\n",
      "|Ford Torino                     |\n",
      "|Ford Galaxie 500                |\n",
      "|Chevrolet Impala                |\n",
      "|Plymouth Fury iii               |\n",
      "|Pontiac Catalina                |\n",
      "|AMC Ambassador DPL              |\n",
      "|Citroen DS-21 Pallas            |\n",
      "|Chevrolet Chevelle Concours (sw)|\n",
      "|Ford Torino (sw)                |\n",
      "|Plymouth Satellite (sw)         |\n",
      "|AMC Rebel SST (sw)              |\n",
      "|Dodge Challenger SE             |\n",
      "|Plymouth 'Cuda 340              |\n",
      "|Ford Mustang Boss 302           |\n",
      "|Chevrolet Monte Carlo           |\n",
      "|Buick Estate Wagon (sw)         |\n",
      "+--------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 1st method\n",
    "# Column name is case sensitive in this usage\n",
    "print(df.Car)\n",
    "print(\"*\"*20)\n",
    "df.select(df.Car).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "YxP1su8veNde"
   },
   "source": [
    "**NOTE:**\n",
    "\n",
    "> **We can't always use the dot notation because this will break when the column names have reserved names or attributes to the data frame class. Additionally, the column names are case sensitive in nature so we need to always make sure the column names have been changed to a paticular case before using it.**\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "md5zaET8dsr4",
    "outputId": "534f2e91-1c8a-48c9-c666-8af90443e896"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Column<'car'>\n",
      "********************\n",
      "+--------------------------------+\n",
      "|car                             |\n",
      "+--------------------------------+\n",
      "|Chevrolet Chevelle Malibu       |\n",
      "|Buick Skylark 320               |\n",
      "|Plymouth Satellite              |\n",
      "|AMC Rebel SST                   |\n",
      "|Ford Torino                     |\n",
      "|Ford Galaxie 500                |\n",
      "|Chevrolet Impala                |\n",
      "|Plymouth Fury iii               |\n",
      "|Pontiac Catalina                |\n",
      "|AMC Ambassador DPL              |\n",
      "|Citroen DS-21 Pallas            |\n",
      "|Chevrolet Chevelle Concours (sw)|\n",
      "|Ford Torino (sw)                |\n",
      "|Plymouth Satellite (sw)         |\n",
      "|AMC Rebel SST (sw)              |\n",
      "|Dodge Challenger SE             |\n",
      "|Plymouth 'Cuda 340              |\n",
      "|Ford Mustang Boss 302           |\n",
      "|Chevrolet Monte Carlo           |\n",
      "|Buick Estate Wagon (sw)         |\n",
      "+--------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 2nd method\n",
    "# Column name is case insensitive here\n",
    "print(df['car'])\n",
    "print(\"*\"*20)\n",
    "df.select(df['car']).show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "6Gkf14sHec9a",
    "outputId": "a965002e-eb65-462d-fa1f-4959251216cd"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+\n",
      "|car                             |\n",
      "+--------------------------------+\n",
      "|Chevrolet Chevelle Malibu       |\n",
      "|Buick Skylark 320               |\n",
      "|Plymouth Satellite              |\n",
      "|AMC Rebel SST                   |\n",
      "|Ford Torino                     |\n",
      "|Ford Galaxie 500                |\n",
      "|Chevrolet Impala                |\n",
      "|Plymouth Fury iii               |\n",
      "|Pontiac Catalina                |\n",
      "|AMC Ambassador DPL              |\n",
      "|Citroen DS-21 Pallas            |\n",
      "|Chevrolet Chevelle Concours (sw)|\n",
      "|Ford Torino (sw)                |\n",
      "|Plymouth Satellite (sw)         |\n",
      "|AMC Rebel SST (sw)              |\n",
      "|Dodge Challenger SE             |\n",
      "|Plymouth 'Cuda 340              |\n",
      "|Ford Mustang Boss 302           |\n",
      "|Chevrolet Monte Carlo           |\n",
      "|Buick Estate Wagon (sw)         |\n",
      "+--------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 3rd method\n",
    "# Column name is case insensitive here\n",
    "from pyspark.sql.functions import col\n",
    "df.select(col('car')).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "z6QsMfnNt3qF"
   },
   "source": [
    "<a id='selecting-multiple-columns'></a>\n",
    "### Selecting Multiple Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "bPjLMhZ6uAQR",
    "outputId": "920f8155-50f5-43ae-eb3d-2b9f6be7756e"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Column<'Car'> Column<'Cylinders'>\n",
      "****************************************\n",
      "+--------------------------------+---------+\n",
      "|Car                             |Cylinders|\n",
      "+--------------------------------+---------+\n",
      "|Chevrolet Chevelle Malibu       |8        |\n",
      "|Buick Skylark 320               |8        |\n",
      "|Plymouth Satellite              |8        |\n",
      "|AMC Rebel SST                   |8        |\n",
      "|Ford Torino                     |8        |\n",
      "|Ford Galaxie 500                |8        |\n",
      "|Chevrolet Impala                |8        |\n",
      "|Plymouth Fury iii               |8        |\n",
      "|Pontiac Catalina                |8        |\n",
      "|AMC Ambassador DPL              |8        |\n",
      "|Citroen DS-21 Pallas            |4        |\n",
      "|Chevrolet Chevelle Concours (sw)|8        |\n",
      "|Ford Torino (sw)                |8        |\n",
      "|Plymouth Satellite (sw)         |8        |\n",
      "|AMC Rebel SST (sw)              |8        |\n",
      "|Dodge Challenger SE             |8        |\n",
      "|Plymouth 'Cuda 340              |8        |\n",
      "|Ford Mustang Boss 302           |8        |\n",
      "|Chevrolet Monte Carlo           |8        |\n",
      "|Buick Estate Wagon (sw)         |8        |\n",
      "+--------------------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 1st method\n",
    "# Column name is case sensitive in this usage\n",
    "print(df.Car, df.Cylinders)\n",
    "print(\"*\"*40)\n",
    "df.select(df.Car, df.Cylinders).show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "DMRMUrv7uHWa",
    "outputId": "711ecef5-9f20-4106-b935-fb722ec14f72"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Column<'car'> Column<'cylinders'>\n",
      "****************************************\n",
      "+--------------------------------+---------+\n",
      "|car                             |cylinders|\n",
      "+--------------------------------+---------+\n",
      "|Chevrolet Chevelle Malibu       |8        |\n",
      "|Buick Skylark 320               |8        |\n",
      "|Plymouth Satellite              |8        |\n",
      "|AMC Rebel SST                   |8        |\n",
      "|Ford Torino                     |8        |\n",
      "|Ford Galaxie 500                |8        |\n",
      "|Chevrolet Impala                |8        |\n",
      "|Plymouth Fury iii               |8        |\n",
      "|Pontiac Catalina                |8        |\n",
      "|AMC Ambassador DPL              |8        |\n",
      "|Citroen DS-21 Pallas            |4        |\n",
      "|Chevrolet Chevelle Concours (sw)|8        |\n",
      "|Ford Torino (sw)                |8        |\n",
      "|Plymouth Satellite (sw)         |8        |\n",
      "|AMC Rebel SST (sw)              |8        |\n",
      "|Dodge Challenger SE             |8        |\n",
      "|Plymouth 'Cuda 340              |8        |\n",
      "|Ford Mustang Boss 302           |8        |\n",
      "|Chevrolet Monte Carlo           |8        |\n",
      "|Buick Estate Wagon (sw)         |8        |\n",
      "+--------------------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 2nd method\n",
    "# Column name is case insensitive in this usage\n",
    "print(df['car'],df['cylinders'])\n",
    "print(\"*\"*40)\n",
    "df.select(df['car'],df['cylinders']).show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "RgQ20-4GugjR",
    "outputId": "91f3f5e6-b90d-413b-ab7f-10efc81b77aa"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+---------+\n",
      "|car                             |cylinders|\n",
      "+--------------------------------+---------+\n",
      "|Chevrolet Chevelle Malibu       |8        |\n",
      "|Buick Skylark 320               |8        |\n",
      "|Plymouth Satellite              |8        |\n",
      "|AMC Rebel SST                   |8        |\n",
      "|Ford Torino                     |8        |\n",
      "|Ford Galaxie 500                |8        |\n",
      "|Chevrolet Impala                |8        |\n",
      "|Plymouth Fury iii               |8        |\n",
      "|Pontiac Catalina                |8        |\n",
      "|AMC Ambassador DPL              |8        |\n",
      "|Citroen DS-21 Pallas            |4        |\n",
      "|Chevrolet Chevelle Concours (sw)|8        |\n",
      "|Ford Torino (sw)                |8        |\n",
      "|Plymouth Satellite (sw)         |8        |\n",
      "|AMC Rebel SST (sw)              |8        |\n",
      "|Dodge Challenger SE             |8        |\n",
      "|Plymouth 'Cuda 340              |8        |\n",
      "|Ford Mustang Boss 302           |8        |\n",
      "|Chevrolet Monte Carlo           |8        |\n",
      "|Buick Estate Wagon (sw)         |8        |\n",
      "+--------------------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 3rd method\n",
    "# Column name is case insensitive in this usage\n",
    "from pyspark.sql.functions import col\n",
    "df.select(col('car'),col('cylinders')).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "85Lv3zSXCcOY"
   },
   "source": [
    "<a id='adding-new-columns'></a>\n",
    "### Adding New Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "d_Y7dcAHu-Uz"
   },
   "source": [
    "We will take a look at three cases here:\n",
    "\n",
    "1.   Adding a new column\n",
    "2.   Adding multiple columns\n",
    "3.   Deriving a new column from an exisitng one"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "oFHUmRKZeCEV",
    "outputId": "4ed01c11-52d6-4ab3-a1ff-0423ef192293"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+\n",
      "|Car                      |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|first_column|\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+\n",
      "|Chevrolet Chevelle Malibu|18.0|8        |307.0       |130.0     |3504.0|12.0        |70   |US    |1           |\n",
      "|Buick Skylark 320        |15.0|8        |350.0       |165.0     |3693.0|11.5        |70   |US    |1           |\n",
      "|Plymouth Satellite       |18.0|8        |318.0       |150.0     |3436.0|11.0        |70   |US    |1           |\n",
      "|AMC Rebel SST            |16.0|8        |304.0       |150.0     |3433.0|12.0        |70   |US    |1           |\n",
      "|Ford Torino              |17.0|8        |302.0       |140.0     |3449.0|10.5        |70   |US    |1           |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# CASE 1: Adding a new column\n",
    "# We will add a new column called 'first_column' at the end\n",
    "from pyspark.sql.functions import lit\n",
    "df = df.withColumn('first_column',lit(1)) \n",
    "# lit means literal. It populates the row with the literal value given.\n",
    "# When adding static data / constant values, it is a good practice to use it.\n",
    "df.show(5,truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "U9772_mHwAqL",
    "outputId": "fe35b15f-cedb-4bda-eaf8-f817c55162e0"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+-------------+------------+\n",
      "|Car                      |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|first_column|second_column|third_column|\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+-------------+------------+\n",
      "|Chevrolet Chevelle Malibu|18.0|8        |307.0       |130.0     |3504.0|12.0        |70   |US    |1           |2            |Third Column|\n",
      "|Buick Skylark 320        |15.0|8        |350.0       |165.0     |3693.0|11.5        |70   |US    |1           |2            |Third Column|\n",
      "|Plymouth Satellite       |18.0|8        |318.0       |150.0     |3436.0|11.0        |70   |US    |1           |2            |Third Column|\n",
      "|AMC Rebel SST            |16.0|8        |304.0       |150.0     |3433.0|12.0        |70   |US    |1           |2            |Third Column|\n",
      "|Ford Torino              |17.0|8        |302.0       |140.0     |3449.0|10.5        |70   |US    |1           |2            |Third Column|\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+-------------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# CASE 2: Adding multiple columns\n",
    "# We will add two new columns called 'second_column' and 'third_column' at the end\n",
    "df = df.withColumn('second_column', lit(2)) \\\n",
    "       .withColumn('third_column', lit('Third Column')) \n",
    "# lit means literal. It populates the row with the literal value given.\n",
    "# When adding static data / constant values, it is a good practice to use it.\n",
    "df.show(5,truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "dGaQS_pOwx_b",
    "outputId": "f21c6b33-4e2a-41bd-cbbc-6b8712de1cbd"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+-------------+------------+----------------------------+\n",
      "|Car                      |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|first_column|second_column|third_column|car_model                   |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+-------------+------------+----------------------------+\n",
      "|Chevrolet Chevelle Malibu|18.0|8        |307.0       |130.0     |3504.0|12.0        |70   |US    |1           |2            |Third Column|Chevrolet Chevelle Malibu 70|\n",
      "|Buick Skylark 320        |15.0|8        |350.0       |165.0     |3693.0|11.5        |70   |US    |1           |2            |Third Column|Buick Skylark 320 70        |\n",
      "|Plymouth Satellite       |18.0|8        |318.0       |150.0     |3436.0|11.0        |70   |US    |1           |2            |Third Column|Plymouth Satellite 70       |\n",
      "|AMC Rebel SST            |16.0|8        |304.0       |150.0     |3433.0|12.0        |70   |US    |1           |2            |Third Column|AMC Rebel SST 70            |\n",
      "|Ford Torino              |17.0|8        |302.0       |140.0     |3449.0|10.5        |70   |US    |1           |2            |Third Column|Ford Torino 70              |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+------------+-------------+------------+----------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# CASE 3: Deriving a new column from an exisitng one\n",
    "# We will add a new column called 'car_model' which has the value of car and model appended together with a space in between \n",
    "from pyspark.sql.functions import concat\n",
    "df = df.withColumn('car_model', concat(col(\"Car\"), lit(\" \"), col(\"model\")))\n",
    "# lit means literal. It populates the row with the literal value given.\n",
    "# When adding static data / constant values, it is a good practice to use it.\n",
    "df.show(5,truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "xeHExg-zxf5r"
   },
   "source": [
    "As we can see, the new column car model has been created from existing columns. Since our aim was to create a column which has the value of car and model appended together with a space in between we have used the `concat` operator."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "QlMf04i2CjDC"
   },
   "source": [
    "<a id='renaming-columns'></a>\n",
    "### Renaming Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "CwGKbSHvxxxG"
   },
   "source": [
    "We use the `withColumnRenamed` function to rename a columm in PySpark. Let us see it in action below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "QJqgy6lKfk2o",
    "outputId": "5abdd234-5373-4cbd-89e0-04312ac6123d"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+----+---------+------------+----------+------+------------+-----+------+--------------+--------------+----------------+-----------------------------------+\n",
      "|Car                             |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|new_column_one|new_column_two|new_column_three|car_model                          |\n",
      "+--------------------------------+----+---------+------------+----------+------+------------+-----+------+--------------+--------------+----------------+-----------------------------------+\n",
      "|Chevrolet Chevelle Malibu       |18.0|8        |307.0       |130.0     |3504.0|12.0        |70   |US    |1             |2             |Third Column    |Chevrolet Chevelle Malibu 70       |\n",
      "|Buick Skylark 320               |15.0|8        |350.0       |165.0     |3693.0|11.5        |70   |US    |1             |2             |Third Column    |Buick Skylark 320 70               |\n",
      "|Plymouth Satellite              |18.0|8        |318.0       |150.0     |3436.0|11.0        |70   |US    |1             |2             |Third Column    |Plymouth Satellite 70              |\n",
      "|AMC Rebel SST                   |16.0|8        |304.0       |150.0     |3433.0|12.0        |70   |US    |1             |2             |Third Column    |AMC Rebel SST 70                   |\n",
      "|Ford Torino                     |17.0|8        |302.0       |140.0     |3449.0|10.5        |70   |US    |1             |2             |Third Column    |Ford Torino 70                     |\n",
      "|Ford Galaxie 500                |15.0|8        |429.0       |198.0     |4341.0|10.0        |70   |US    |1             |2             |Third Column    |Ford Galaxie 500 70                |\n",
      "|Chevrolet Impala                |14.0|8        |454.0       |220.0     |4354.0|9.0         |70   |US    |1             |2             |Third Column    |Chevrolet Impala 70                |\n",
      "|Plymouth Fury iii               |14.0|8        |440.0       |215.0     |4312.0|8.5         |70   |US    |1             |2             |Third Column    |Plymouth Fury iii 70               |\n",
      "|Pontiac Catalina                |14.0|8        |455.0       |225.0     |4425.0|10.0        |70   |US    |1             |2             |Third Column    |Pontiac Catalina 70                |\n",
      "|AMC Ambassador DPL              |15.0|8        |390.0       |190.0     |3850.0|8.5         |70   |US    |1             |2             |Third Column    |AMC Ambassador DPL 70              |\n",
      "|Citroen DS-21 Pallas            |0.0 |4        |133.0       |115.0     |3090.0|17.5        |70   |Europe|1             |2             |Third Column    |Citroen DS-21 Pallas 70            |\n",
      "|Chevrolet Chevelle Concours (sw)|0.0 |8        |350.0       |165.0     |4142.0|11.5        |70   |US    |1             |2             |Third Column    |Chevrolet Chevelle Concours (sw) 70|\n",
      "|Ford Torino (sw)                |0.0 |8        |351.0       |153.0     |4034.0|11.0        |70   |US    |1             |2             |Third Column    |Ford Torino (sw) 70                |\n",
      "|Plymouth Satellite (sw)         |0.0 |8        |383.0       |175.0     |4166.0|10.5        |70   |US    |1             |2             |Third Column    |Plymouth Satellite (sw) 70         |\n",
      "|AMC Rebel SST (sw)              |0.0 |8        |360.0       |175.0     |3850.0|11.0        |70   |US    |1             |2             |Third Column    |AMC Rebel SST (sw) 70              |\n",
      "|Dodge Challenger SE             |15.0|8        |383.0       |170.0     |3563.0|10.0        |70   |US    |1             |2             |Third Column    |Dodge Challenger SE 70             |\n",
      "|Plymouth 'Cuda 340              |14.0|8        |340.0       |160.0     |3609.0|8.0         |70   |US    |1             |2             |Third Column    |Plymouth 'Cuda 340 70              |\n",
      "|Ford Mustang Boss 302           |0.0 |8        |302.0       |140.0     |3353.0|8.0         |70   |US    |1             |2             |Third Column    |Ford Mustang Boss 302 70           |\n",
      "|Chevrolet Monte Carlo           |15.0|8        |400.0       |150.0     |3761.0|9.5         |70   |US    |1             |2             |Third Column    |Chevrolet Monte Carlo 70           |\n",
      "|Buick Estate Wagon (sw)         |14.0|8        |455.0       |225.0     |3086.0|10.0        |70   |US    |1             |2             |Third Column    |Buick Estate Wagon (sw) 70         |\n",
      "+--------------------------------+----+---------+------------+----------+------+------------+-----+------+--------------+--------------+----------------+-----------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Renaming a column in PySpark\n",
    "df = df.withColumnRenamed('first_column', 'new_column_one') \\\n",
    "       .withColumnRenamed('second_column', 'new_column_two') \\\n",
    "       .withColumnRenamed('third_column', 'new_column_three')\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "4CDifVC2Cnml"
   },
   "source": [
    "<a id='grouping-by-columns'></a>\n",
    "### Grouping By Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "9wlB76FdyS0W"
   },
   "source": [
    "Here, we see the Dataframe API way of grouping values. We will discuss how to:\n",
    "\n",
    "\n",
    "1.   Group By a single column\n",
    "2.   Group By multiple columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "M1ek2opVfqea",
    "outputId": "3eab0db5-13ab-4868-871b-da4fb16ab768"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|Origin|count|\n",
      "+------+-----+\n",
      "|Europe|   73|\n",
      "|    US|  254|\n",
      "| Japan|   79|\n",
      "+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Group By a column in PySpark\n",
    "df.groupBy('Origin').count().show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "hUh_TWcOysoL",
    "outputId": "dad5b95e-4949-4259-cbd4-b0103d6daec9"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+-----+\n",
      "|Origin|Model|count|\n",
      "+------+-----+-----+\n",
      "|Europe|   71|    5|\n",
      "|Europe|   80|    9|\n",
      "|Europe|   79|    4|\n",
      "| Japan|   75|    4|\n",
      "|    US|   72|   18|\n",
      "+------+-----+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Group By multiple columns in PySpark\n",
    "df.groupBy('Origin', 'Model').count().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "CbpEj9fECrW3"
   },
   "source": [
    "<a id='removing-columns'></a>\n",
    "### Removing Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "xsb9PXxpfnmh",
    "outputId": "601341ee-c131-46e3-9331-2e93664f9bb2"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+--------------+----------------+----------------------------+\n",
      "|Car                      |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|new_column_two|new_column_three|car_model                   |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+--------------+----------------+----------------------------+\n",
      "|Chevrolet Chevelle Malibu|18.0|8        |307.0       |130.0     |3504.0|12.0        |70   |US    |2             |Third Column    |Chevrolet Chevelle Malibu 70|\n",
      "|Buick Skylark 320        |15.0|8        |350.0       |165.0     |3693.0|11.5        |70   |US    |2             |Third Column    |Buick Skylark 320 70        |\n",
      "|Plymouth Satellite       |18.0|8        |318.0       |150.0     |3436.0|11.0        |70   |US    |2             |Third Column    |Plymouth Satellite 70       |\n",
      "|AMC Rebel SST            |16.0|8        |304.0       |150.0     |3433.0|12.0        |70   |US    |2             |Third Column    |AMC Rebel SST 70            |\n",
      "|Ford Torino              |17.0|8        |302.0       |140.0     |3449.0|10.5        |70   |US    |2             |Third Column    |Ford Torino 70              |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+--------------+----------------+----------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Remove columns in PySpark\n",
    "df = df.drop('new_column_one')\n",
    "df.show(5,truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "EKOXrXtvzK_0",
    "outputId": "329e9696-7a60-4fcf-dc56-4d6cbcd5dc7a"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+----------------------------+\n",
      "|Car                      |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|car_model                   |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+----------------------------+\n",
      "|Chevrolet Chevelle Malibu|18.0|8        |307.0       |130.0     |3504.0|12.0        |70   |US    |Chevrolet Chevelle Malibu 70|\n",
      "|Buick Skylark 320        |15.0|8        |350.0       |165.0     |3693.0|11.5        |70   |US    |Buick Skylark 320 70        |\n",
      "|Plymouth Satellite       |18.0|8        |318.0       |150.0     |3436.0|11.0        |70   |US    |Plymouth Satellite 70       |\n",
      "|AMC Rebel SST            |16.0|8        |304.0       |150.0     |3433.0|12.0        |70   |US    |AMC Rebel SST 70            |\n",
      "|Ford Torino              |17.0|8        |302.0       |140.0     |3449.0|10.5        |70   |US    |Ford Torino 70              |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+----------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Remove multiple columnss in one go\n",
    "df = df.drop('new_column_two') \\\n",
    "       .drop('new_column_three')\n",
    "df.show(5,truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "WbKK5iHwmIoV"
   },
   "source": [
    "<a id='dataframe-operations-on-rows'></a>\n",
    "## DataFrame Operations on Rows"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Quwx3KlLzeq9"
   },
   "source": [
    "We will discuss the follwoing in this section:\n",
    "\n",
    "1.   Filtering Rows\n",
    "2. \t Get Distinct Rows\n",
    "3.   Sorting Rows\n",
    "4.   Union Dataframes\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "9bKlvX-SH-Wy"
   },
   "source": [
    "<a id='filtering-rows'></a>\n",
    "### Filtering Rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "YNfcjOIknA3n",
    "outputId": "7319fc9b-76cb-428f-dbb6-99f25b80b79c"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "TOTAL RECORD COUNT: 406\n",
      "EUROPE FILTERED RECORD COUNT: 73\n",
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "|Car                         |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|car_model                      |\n",
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "|Citroen DS-21 Pallas        |0.0 |4        |133.0       |115.0     |3090.0|17.5        |70   |Europe|Citroen DS-21 Pallas 70        |\n",
      "|Volkswagen 1131 Deluxe Sedan|26.0|4        |97.0        |46.0      |1835.0|20.5        |70   |Europe|Volkswagen 1131 Deluxe Sedan 70|\n",
      "|Peugeot 504                 |25.0|4        |110.0       |87.0      |2672.0|17.5        |70   |Europe|Peugeot 504 70                 |\n",
      "|Audi 100 LS                 |24.0|4        |107.0       |90.0      |2430.0|14.5        |70   |Europe|Audi 100 LS 70                 |\n",
      "|Saab 99e                    |25.0|4        |104.0       |95.0      |2375.0|17.5        |70   |Europe|Saab 99e 70                    |\n",
      "|BMW 2002                    |26.0|4        |121.0       |113.0     |2234.0|12.5        |70   |Europe|BMW 2002 70                    |\n",
      "|Volkswagen Super Beetle 117 |0.0 |4        |97.0        |48.0      |1978.0|20.0        |71   |Europe|Volkswagen Super Beetle 117 71 |\n",
      "|Opel 1900                   |28.0|4        |116.0       |90.0      |2123.0|14.0        |71   |Europe|Opel 1900 71                   |\n",
      "|Peugeot 304                 |30.0|4        |79.0        |70.0      |2074.0|19.5        |71   |Europe|Peugeot 304 71                 |\n",
      "|Fiat 124B                   |30.0|4        |88.0        |76.0      |2065.0|14.5        |71   |Europe|Fiat 124B 71                   |\n",
      "|Volkswagen Model 111        |27.0|4        |97.0        |60.0      |1834.0|19.0        |71   |Europe|Volkswagen Model 111 71        |\n",
      "|Volkswagen Type 3           |23.0|4        |97.0        |54.0      |2254.0|23.5        |72   |Europe|Volkswagen Type 3 72           |\n",
      "|Volvo 145e (sw)             |18.0|4        |121.0       |112.0     |2933.0|14.5        |72   |Europe|Volvo 145e (sw) 72             |\n",
      "|Volkswagen 411 (sw)         |22.0|4        |121.0       |76.0      |2511.0|18.0        |72   |Europe|Volkswagen 411 (sw) 72         |\n",
      "|Peugeot 504 (sw)            |21.0|4        |120.0       |87.0      |2979.0|19.5        |72   |Europe|Peugeot 504 (sw) 72            |\n",
      "|Renault 12 (sw)             |26.0|4        |96.0        |69.0      |2189.0|18.0        |72   |Europe|Renault 12 (sw) 72             |\n",
      "|Volkswagen Super Beetle     |26.0|4        |97.0        |46.0      |1950.0|21.0        |73   |Europe|Volkswagen Super Beetle 73     |\n",
      "|Fiat 124 Sport Coupe        |26.0|4        |98.0        |90.0      |2265.0|15.5        |73   |Europe|Fiat 124 Sport Coupe 73        |\n",
      "|Fiat 128                    |29.0|4        |68.0        |49.0      |1867.0|19.5        |73   |Europe|Fiat 128 73                    |\n",
      "|Opel Manta                  |24.0|4        |116.0       |75.0      |2158.0|15.5        |73   |Europe|Opel Manta 73                  |\n",
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filtering rows in PySpark\n",
    "total_count = df.count()\n",
    "print(\"TOTAL RECORD COUNT: \" + str(total_count)) \n",
    "europe_filtered_count = df.filter(col('Origin')=='Europe').count()\n",
    "print(\"EUROPE FILTERED RECORD COUNT: \" + str(europe_filtered_count))\n",
    "df.filter(col('Origin')=='Europe').show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "MXJxRwBQ1lyd",
    "outputId": "06fcc49c-c21e-47c5-ee14-324ae724e274"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "TOTAL RECORD COUNT: 406\n",
      "EUROPE FILTERED RECORD COUNT: 66\n",
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "|Car                         |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|car_model                      |\n",
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "|Citroen DS-21 Pallas        |0.0 |4        |133.0       |115.0     |3090.0|17.5        |70   |Europe|Citroen DS-21 Pallas 70        |\n",
      "|Volkswagen 1131 Deluxe Sedan|26.0|4        |97.0        |46.0      |1835.0|20.5        |70   |Europe|Volkswagen 1131 Deluxe Sedan 70|\n",
      "|Peugeot 504                 |25.0|4        |110.0       |87.0      |2672.0|17.5        |70   |Europe|Peugeot 504 70                 |\n",
      "|Audi 100 LS                 |24.0|4        |107.0       |90.0      |2430.0|14.5        |70   |Europe|Audi 100 LS 70                 |\n",
      "|Saab 99e                    |25.0|4        |104.0       |95.0      |2375.0|17.5        |70   |Europe|Saab 99e 70                    |\n",
      "|BMW 2002                    |26.0|4        |121.0       |113.0     |2234.0|12.5        |70   |Europe|BMW 2002 70                    |\n",
      "|Volkswagen Super Beetle 117 |0.0 |4        |97.0        |48.0      |1978.0|20.0        |71   |Europe|Volkswagen Super Beetle 117 71 |\n",
      "|Opel 1900                   |28.0|4        |116.0       |90.0      |2123.0|14.0        |71   |Europe|Opel 1900 71                   |\n",
      "|Peugeot 304                 |30.0|4        |79.0        |70.0      |2074.0|19.5        |71   |Europe|Peugeot 304 71                 |\n",
      "|Fiat 124B                   |30.0|4        |88.0        |76.0      |2065.0|14.5        |71   |Europe|Fiat 124B 71                   |\n",
      "|Volkswagen Model 111        |27.0|4        |97.0        |60.0      |1834.0|19.0        |71   |Europe|Volkswagen Model 111 71        |\n",
      "|Volkswagen Type 3           |23.0|4        |97.0        |54.0      |2254.0|23.5        |72   |Europe|Volkswagen Type 3 72           |\n",
      "|Volvo 145e (sw)             |18.0|4        |121.0       |112.0     |2933.0|14.5        |72   |Europe|Volvo 145e (sw) 72             |\n",
      "|Volkswagen 411 (sw)         |22.0|4        |121.0       |76.0      |2511.0|18.0        |72   |Europe|Volkswagen 411 (sw) 72         |\n",
      "|Peugeot 504 (sw)            |21.0|4        |120.0       |87.0      |2979.0|19.5        |72   |Europe|Peugeot 504 (sw) 72            |\n",
      "|Renault 12 (sw)             |26.0|4        |96.0        |69.0      |2189.0|18.0        |72   |Europe|Renault 12 (sw) 72             |\n",
      "|Volkswagen Super Beetle     |26.0|4        |97.0        |46.0      |1950.0|21.0        |73   |Europe|Volkswagen Super Beetle 73     |\n",
      "|Fiat 124 Sport Coupe        |26.0|4        |98.0        |90.0      |2265.0|15.5        |73   |Europe|Fiat 124 Sport Coupe 73        |\n",
      "|Fiat 128                    |29.0|4        |68.0        |49.0      |1867.0|19.5        |73   |Europe|Fiat 128 73                    |\n",
      "|Opel Manta                  |24.0|4        |116.0       |75.0      |2158.0|15.5        |73   |Europe|Opel Manta 73                  |\n",
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filtering rows in PySpark based on Multiple conditions\n",
    "total_count = df.count()\n",
    "print(\"TOTAL RECORD COUNT: \" + str(total_count)) \n",
    "europe_filtered_count = df.filter((col('Origin')=='Europe') & \n",
    "                                  (col('Cylinders')==4)).count() # Two conditions added here\n",
    "print(\"EUROPE FILTERED RECORD COUNT: \" + str(europe_filtered_count))\n",
    "df.filter(col('Origin')=='Europe').show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "zLU-a4auIEvh"
   },
   "source": [
    "<a id='get-distinct-rows'></a>\n",
    "### Get Distinct Rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "B1RKg1UrmBQz",
    "outputId": "d2713c65-a75b-4b64-b6e1-181b96259649"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|Origin|\n",
      "+------+\n",
      "|Europe|\n",
      "|    US|\n",
      "| Japan|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Get Unique Rows in PySpark\n",
    "df.select('Origin').distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "_LQWXPXt0g0N",
    "outputId": "6098ccd8-ae6d-4772-b18a-aad49f8d58c0"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|Origin|model|\n",
      "+------+-----+\n",
      "|Europe|   71|\n",
      "|Europe|   80|\n",
      "|Europe|   79|\n",
      "| Japan|   75|\n",
      "|    US|   72|\n",
      "|    US|   80|\n",
      "|Europe|   74|\n",
      "| Japan|   79|\n",
      "|Europe|   76|\n",
      "|    US|   75|\n",
      "| Japan|   77|\n",
      "|    US|   82|\n",
      "| Japan|   80|\n",
      "| Japan|   78|\n",
      "|    US|   78|\n",
      "|Europe|   75|\n",
      "|    US|   71|\n",
      "|    US|   77|\n",
      "| Japan|   70|\n",
      "| Japan|   71|\n",
      "+------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Get Unique Rows in PySpark based on mutliple columns\n",
    "df.select('Origin','model').distinct().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "-069UYUwIIYI"
   },
   "source": [
    "<a id='sorting-rows'></a>\n",
    "### Sorting Rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "4ZpeJvz0nkBI",
    "outputId": "fa965b71-521e-48b7-b440-0033a4ae6599"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "|Car                         |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|car_model                      |\n",
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "|Mazda RX-4                  |21.5|3        |80.0        |110.0     |2720.0|13.5        |77   |Japan |Mazda RX-4 77                  |\n",
      "|Mazda RX-7 GS               |23.7|3        |70.0        |100.0     |2420.0|12.5        |80   |Japan |Mazda RX-7 GS 80               |\n",
      "|Mazda RX2 Coupe             |19.0|3        |70.0        |97.0      |2330.0|13.5        |72   |Japan |Mazda RX2 Coupe 72             |\n",
      "|Mazda RX3                   |18.0|3        |70.0        |90.0      |2124.0|13.5        |73   |Japan |Mazda RX3 73                   |\n",
      "|Datsun 510 (sw)             |28.0|4        |97.0        |92.0      |2288.0|17.0        |72   |Japan |Datsun 510 (sw) 72             |\n",
      "|Opel 1900                   |28.0|4        |116.0       |90.0      |2123.0|14.0        |71   |Europe|Opel 1900 71                   |\n",
      "|Mercury Capri 2000          |23.0|4        |122.0       |86.0      |2220.0|14.0        |71   |US    |Mercury Capri 2000 71          |\n",
      "|Volkswagen 1131 Deluxe Sedan|26.0|4        |97.0        |46.0      |1835.0|20.5        |70   |Europe|Volkswagen 1131 Deluxe Sedan 70|\n",
      "|Peugeot 304                 |30.0|4        |79.0        |70.0      |2074.0|19.5        |71   |Europe|Peugeot 304 71                 |\n",
      "|Fiat 124B                   |30.0|4        |88.0        |76.0      |2065.0|14.5        |71   |Europe|Fiat 124B 71                   |\n",
      "|Chevrolet Vega (sw)         |22.0|4        |140.0       |72.0      |2408.0|19.0        |71   |US    |Chevrolet Vega (sw) 71         |\n",
      "|Datsun 1200                 |35.0|4        |72.0        |69.0      |1613.0|18.0        |71   |Japan |Datsun 1200 71                 |\n",
      "|Volkswagen Model 111        |27.0|4        |97.0        |60.0      |1834.0|19.0        |71   |Europe|Volkswagen Model 111 71        |\n",
      "|Volkswagen Type 3           |23.0|4        |97.0        |54.0      |2254.0|23.5        |72   |Europe|Volkswagen Type 3 72           |\n",
      "|Audi 100 LS                 |24.0|4        |107.0       |90.0      |2430.0|14.5        |70   |Europe|Audi 100 LS 70                 |\n",
      "|BMW 2002                    |26.0|4        |121.0       |113.0     |2234.0|12.5        |70   |Europe|BMW 2002 70                    |\n",
      "|Toyota Corolla 1200         |31.0|4        |71.0        |65.0      |1773.0|19.0        |71   |Japan |Toyota Corolla 1200 71         |\n",
      "|Chevrolet Vega 2300         |28.0|4        |140.0       |90.0      |2264.0|15.5        |71   |US    |Chevrolet Vega 2300 71         |\n",
      "|Ford Pinto                  |25.0|4        |98.0        |0.0       |2046.0|19.0        |71   |US    |Ford Pinto 71                  |\n",
      "|Dodge Colt Hardtop          |25.0|4        |97.5        |80.0      |2126.0|17.0        |72   |US    |Dodge Colt Hardtop 72          |\n",
      "+----------------------------+----+---------+------------+----------+------+------------+-----+------+-------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Sort Rows in PySpark\n",
    "# By default the data will be sorted in ascending order\n",
    "df.orderBy('Cylinders').show(truncate=False) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "v1CEwofMJV-D",
    "outputId": "46809931-e309-4482-adf1-2bd7cfdac811"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+----------------------------+\n",
      "|Car                      |MPG |Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|car_model                   |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+----------------------------+\n",
      "|Plymouth 'Cuda 340       |14.0|8        |340.0       |160.0     |3609.0|8.0         |70   |US    |Plymouth 'Cuda 340 70       |\n",
      "|Pontiac Safari (sw)      |13.0|8        |400.0       |175.0     |5140.0|12.0        |71   |US    |Pontiac Safari (sw) 71      |\n",
      "|Ford Mustang Boss 302    |0.0 |8        |302.0       |140.0     |3353.0|8.0         |70   |US    |Ford Mustang Boss 302 70    |\n",
      "|Buick Skylark 320        |15.0|8        |350.0       |165.0     |3693.0|11.5        |70   |US    |Buick Skylark 320 70        |\n",
      "|Chevrolet Monte Carlo    |15.0|8        |400.0       |150.0     |3761.0|9.5         |70   |US    |Chevrolet Monte Carlo 70    |\n",
      "|AMC Rebel SST            |16.0|8        |304.0       |150.0     |3433.0|12.0        |70   |US    |AMC Rebel SST 70            |\n",
      "|Buick Estate Wagon (sw)  |14.0|8        |455.0       |225.0     |3086.0|10.0        |70   |US    |Buick Estate Wagon (sw) 70  |\n",
      "|Ford Galaxie 500         |15.0|8        |429.0       |198.0     |4341.0|10.0        |70   |US    |Ford Galaxie 500 70         |\n",
      "|Ford F250                |10.0|8        |360.0       |215.0     |4615.0|14.0        |70   |US    |Ford F250 70                |\n",
      "|Plymouth Fury iii        |14.0|8        |440.0       |215.0     |4312.0|8.5         |70   |US    |Plymouth Fury iii 70        |\n",
      "|Chevy C20                |10.0|8        |307.0       |200.0     |4376.0|15.0        |70   |US    |Chevy C20 70                |\n",
      "|AMC Ambassador DPL       |15.0|8        |390.0       |190.0     |3850.0|8.5         |70   |US    |AMC Ambassador DPL 70       |\n",
      "|Dodge D200               |11.0|8        |318.0       |210.0     |4382.0|13.5        |70   |US    |Dodge D200 70               |\n",
      "|Ford Torino (sw)         |0.0 |8        |351.0       |153.0     |4034.0|11.0        |70   |US    |Ford Torino (sw) 70         |\n",
      "|Hi 1200D                 |9.0 |8        |304.0       |193.0     |4732.0|18.5        |70   |US    |Hi 1200D 70                 |\n",
      "|AMC Rebel SST (sw)       |0.0 |8        |360.0       |175.0     |3850.0|11.0        |70   |US    |AMC Rebel SST (sw) 70       |\n",
      "|Chevrolet Impala         |14.0|8        |350.0       |165.0     |4209.0|12.0        |71   |US    |Chevrolet Impala 71         |\n",
      "|Chevrolet Chevelle Malibu|18.0|8        |307.0       |130.0     |3504.0|12.0        |70   |US    |Chevrolet Chevelle Malibu 70|\n",
      "|Pontiac Catalina Brougham|14.0|8        |400.0       |175.0     |4464.0|11.5        |71   |US    |Pontiac Catalina Brougham 71|\n",
      "|Ford Torino              |17.0|8        |302.0       |140.0     |3449.0|10.5        |70   |US    |Ford Torino 70              |\n",
      "+-------------------------+----+---------+------------+----------+------+------------+-----+------+----------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# To change the sorting order, you can use the ascending parameter\n",
    "df.orderBy('Cylinders', ascending=False).show(truncate=False) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Zx3W4aeL5A4O",
    "outputId": "6671ffa3-9024-4a5f-9342-2cba4483e387"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|Origin|count|\n",
      "+------+-----+\n",
      "|    US|  254|\n",
      "| Japan|   79|\n",
      "|Europe|   73|\n",
      "+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using groupBy aand orderBy together\n",
    "df.groupBy(\"Origin\").count().orderBy('count', ascending=False).show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "aN0-A_JsIX-X"
   },
   "source": [
    "<a id='union-dataframes'></a>\n",
    "### Union Dataframes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "VH0KOaBrJt6v"
   },
   "source": [
    "You will see three main methods for performing union of dataframes. It is important to know the difference between them and which one is preferred:\n",
    "\n",
    "*   `union()` – It is used to merge two DataFrames of the same structure/schema. If schemas are not the same, it returns an error\n",
    "*   `unionAll()` – This function is deprecated since Spark 2.0.0, and replaced with union()\n",
    "*   `unionByName()` - This function is used to merge two dataframes based on column name.\n",
    "\n",
    "> Since `unionAll()` is deprecated, **`union()` is the preferred method for merging dataframes.**\n",
    "<br>\n",
    "> The difference between `unionByName()` and `union()` is that `unionByName()` resolves columns by name, not by position.\n",
    "\n",
    "In other SQLs, Union eliminates the duplicates but UnionAll merges two datasets, thereby including duplicate records. But, in PySpark, both behave the same and includes duplicate records. The recommendation is to use `distinct()` or `dropDuplicates()` to remove duplicate records."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "bCZIzfYmnx--",
    "outputId": "cd0de201-5f82-4ce3-e15c-35b7665ad079"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "EUROPE CARS: 3\n",
      "JAPAN CARS: 4\n",
      "AFTER UNION: 7\n"
     ]
    }
   ],
   "source": [
    "# CASE 1: Union When columns are in order\n",
    "df = spark.read.csv('cars.csv', header=True, sep=\";\", inferSchema=True)\n",
    "europe_cars = df.filter((col('Origin')=='Europe') & (col('Cylinders')==5))\n",
    "japan_cars = df.filter((col('Origin')=='Japan') & (col('Cylinders')==3))\n",
    "print(\"EUROPE CARS: \"+str(europe_cars.count()))\n",
    "print(\"JAPAN CARS: \"+str(japan_cars.count()))\n",
    "print(\"AFTER UNION: \"+str(europe_cars.union(japan_cars).count()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "1pfPzVOFqC_8"
   },
   "source": [
    "**Result:**\n",
    "\n",
    "> As you can see here, there were 3 cars from Europe with 5 Cylinders, and 4 cars from Japan with 3 Cylinders. After union, there are 7 cars in total.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "CjWjzWBoMxx0",
    "outputId": "11237d45-cacc-403c-e971-6e2717a32135"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+----+\n",
      "|col0|col1|col2|\n",
      "+----+----+----+\n",
      "|   1|   2|   3|\n",
      "|   6|   4|   5|\n",
      "+----+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# CASE 1: Union When columns are not in order\n",
    "# Creating two dataframes with jumbled columns\n",
    "df1 = spark.createDataFrame([[1, 2, 3]], [\"col0\", \"col1\", \"col2\"])\n",
    "df2 = spark.createDataFrame([[4, 5, 6]], [\"col1\", \"col2\", \"col0\"])\n",
    "df1.unionByName(df2).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "EX33t8e3PyGy"
   },
   "source": [
    "**Result:**\n",
    "\n",
    "> As you can see here, the two dataframes have been successfully merged based on their column names.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "aHjILb1DriuX"
   },
   "source": [
    "<a id='common-data-manipulation-functions'></a>\n",
    "## Common Data Manipulation Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "x3vlC7ZerlKb",
    "outputId": "89d386c3-e9c2-4532-c086-f8b00cc38881"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['Column', 'DataFrame', 'DataType', 'PandasUDFType', 'PythonEvalType', 'SparkContext', 'StringType', 'UserDefinedFunction', '__builtins__', '__cached__', '__doc__', '__file__', '__loader__', '__name__', '__package__', '__spec__', '_create_column_from_literal', '_create_lambda', '_create_udf', '_get_get_jvm_function', '_get_lambda_parameters', '_invoke_binary_math_function', '_invoke_function', '_invoke_function_over_column', '_invoke_higher_order_function', '_options_to_str', '_test', '_to_java_column', '_to_seq', '_unresolved_named_lambda_variable', 'abs', 'acos', 'acosh', 'add_months', 'aggregate', 'approxCountDistinct', 'approx_count_distinct', 'array', 'array_contains', 'array_distinct', 'array_except', 'array_intersect', 'array_join', 'array_max', 'array_min', 'array_position', 'array_remove', 'array_repeat', 'array_sort', 'array_union', 'arrays_overlap', 'arrays_zip', 'asc', 'asc_nulls_first', 'asc_nulls_last', 'ascii', 'asin', 'asinh', 'assert_true', 'atan', 'atan2', 'atanh', 'avg', 'base64', 'bin', 'bitwiseNOT', 'broadcast', 'bround', 'bucket', 'cbrt', 'ceil', 'coalesce', 'col', 'collect_list', 'collect_set', 'column', 'concat', 'concat_ws', 'conv', 'corr', 'cos', 'cosh', 'count', 'countDistinct', 'covar_pop', 'covar_samp', 'crc32', 'create_map', 'cume_dist', 'current_date', 'current_timestamp', 'date_add', 'date_format', 'date_sub', 'date_trunc', 'datediff', 'dayofmonth', 'dayofweek', 'dayofyear', 'days', 'decode', 'degrees', 'dense_rank', 'desc', 'desc_nulls_first', 'desc_nulls_last', 'element_at', 'encode', 'exists', 'exp', 'explode', 'explode_outer', 'expm1', 'expr', 'factorial', 'filter', 'first', 'flatten', 'floor', 'forall', 'format_number', 'format_string', 'from_csv', 'from_json', 'from_unixtime', 'from_utc_timestamp', 'functools', 'get_json_object', 'greatest', 'grouping', 'grouping_id', 'hash', 'hex', 'hour', 'hours', 'hypot', 'initcap', 'input_file_name', 'instr', 'isnan', 'isnull', 'json_tuple', 'kurtosis', 'lag', 'last', 'last_day', 'lead', 'least', 'length', 'levenshtein', 'lit', 'locate', 'log', 'log10', 'log1p', 'log2', 'lower', 'lpad', 'ltrim', 'map_concat', 'map_entries', 'map_filter', 'map_from_arrays', 'map_from_entries', 'map_keys', 'map_values', 'map_zip_with', 'max', 'md5', 'mean', 'min', 'minute', 'monotonically_increasing_id', 'month', 'months', 'months_between', 'nanvl', 'next_day', 'nth_value', 'ntile', 'overlay', 'pandas_udf', 'percent_rank', 'percentile_approx', 'posexplode', 'posexplode_outer', 'pow', 'quarter', 'radians', 'raise_error', 'rand', 'randn', 'rank', 'regexp_extract', 'regexp_replace', 'repeat', 'reverse', 'rint', 'round', 'row_number', 'rpad', 'rtrim', 'schema_of_csv', 'schema_of_json', 'second', 'sequence', 'sha1', 'sha2', 'shiftLeft', 'shiftRight', 'shiftRightUnsigned', 'shuffle', 'signum', 'sin', 'since', 'sinh', 'size', 'skewness', 'slice', 'sort_array', 'soundex', 'spark_partition_id', 'split', 'sqrt', 'stddev', 'stddev_pop', 'stddev_samp', 'struct', 'substring', 'substring_index', 'sum', 'sumDistinct', 'sys', 'tan', 'tanh', 'timestamp_seconds', 'toDegrees', 'toRadians', 'to_csv', 'to_date', 'to_json', 'to_str', 'to_timestamp', 'to_utc_timestamp', 'transform', 'transform_keys', 'transform_values', 'translate', 'trim', 'trunc', 'udf', 'unbase64', 'unhex', 'unix_timestamp', 'upper', 'var_pop', 'var_samp', 'variance', 'warnings', 'weekofyear', 'when', 'window', 'xxhash64', 'year', 'years', 'zip_with']\n"
     ]
    }
   ],
   "source": [
    "# Functions available in PySpark\n",
    "from pyspark.sql import functions\n",
    "# Similar to python, we can use the dir function to view the avaiable functions\n",
    "print(dir(functions)) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "PIKigra7A34e"
   },
   "source": [
    "<a id='string-functions'></a>\n",
    "### String Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "id": "63QDccSjBqC4"
   },
   "outputs": [],
   "source": [
    "# Loading the data\n",
    "from pyspark.sql.functions import col\n",
    "df = spark.read.csv('cars.csv', header=True, sep=\";\", inferSchema=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "LiXWN8DUA9x6"
   },
   "source": [
    "**Display the Car column in exisitng, lower and upper characters, and the first 4 characters of the column**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "52Gh9c99BZFr",
    "outputId": "51c7cfa5-7363-4387-cb35-57ee79e22055"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on function substring in module pyspark.sql.functions:\n",
      "\n",
      "substring(str, pos, len)\n",
      "    Substring starts at `pos` and is of length `len` when str is String type or\n",
      "    returns the slice of byte array that starts at `pos` in byte and is of length `len`\n",
      "    when str is Binary type.\n",
      "    \n",
      "    .. versionadded:: 1.5.0\n",
      "    \n",
      "    Notes\n",
      "    -----\n",
      "    The position is not zero based, but 1 based index.\n",
      "    \n",
      "    Examples\n",
      "    --------\n",
      "    >>> df = spark.createDataFrame([('abcd',)], ['s',])\n",
      "    >>> df.select(substring(df.s, 1, 2).alias('s')).collect()\n",
      "    [Row(s='ab')]\n",
      "\n",
      "+-------------------------+-------------------------+-------------------------+------------------+\n",
      "|Car                      |lower(Car)               |upper(Car)               |concatenated value|\n",
      "+-------------------------+-------------------------+-------------------------+------------------+\n",
      "|Chevrolet Chevelle Malibu|chevrolet chevelle malibu|CHEVROLET CHEVELLE MALIBU|Chev              |\n",
      "|Buick Skylark 320        |buick skylark 320        |BUICK SKYLARK 320        |Buic              |\n",
      "|Plymouth Satellite       |plymouth satellite       |PLYMOUTH SATELLITE       |Plym              |\n",
      "|AMC Rebel SST            |amc rebel sst            |AMC REBEL SST            |AMC               |\n",
      "|Ford Torino              |ford torino              |FORD TORINO              |Ford              |\n",
      "+-------------------------+-------------------------+-------------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import col,lower, upper, substring\n",
    "# Prints out the details of a function\n",
    "help(substring)\n",
    "# alias is used to rename the column name in the output\n",
    "df.select(col('Car'),lower(col('Car')),upper(col('Car')),substring(col('Car'),1,4).alias(\"concatenated value\")).show(5, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "GZJFdTbk6rBt"
   },
   "source": [
    "**Concatenate the Car column and Model column and add a space between them.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "8Lo951Cg6phi",
    "outputId": "e3921f2d-9a96-4a33-f397-c458ec391bf5"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+-----+----------------------------+\n",
      "|Car                      |model|concat(Car,  , model)       |\n",
      "+-------------------------+-----+----------------------------+\n",
      "|Chevrolet Chevelle Malibu|70   |Chevrolet Chevelle Malibu 70|\n",
      "|Buick Skylark 320        |70   |Buick Skylark 320 70        |\n",
      "|Plymouth Satellite       |70   |Plymouth Satellite 70       |\n",
      "|AMC Rebel SST            |70   |AMC Rebel SST 70            |\n",
      "|Ford Torino              |70   |Ford Torino 70              |\n",
      "+-------------------------+-----+----------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import concat\n",
    "df.select(col(\"Car\"),col(\"model\"),concat(col(\"Car\"), lit(\" \"), col(\"model\"))).show(5, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ldtA0wk9BMkT"
   },
   "source": [
    "<a id='numeric-functions'></a>\n",
    "### Numeric functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "kmz4G5LVBOs6"
   },
   "source": [
    "**Show the oldest date and the most recent date**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "wBDDH-YpBbdk",
    "outputId": "71d15146-4d47-45d4-af92-f536684fc7b3"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------+\n",
      "|min(Weight)|max(Weight)|\n",
      "+-----------+-----------+\n",
      "|       1613|       5140|\n",
      "+-----------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import min, max\n",
    "df.select(min(col('Weight')), max(col('Weight'))).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "MTg-Royz7Nvi"
   },
   "source": [
    "**Add 10 to the minimum and maximum weight**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "YeiemMsI7Vm2",
    "outputId": "21d6646c-8e6e-47af-bef9-b92ee07c7df7"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+------------------+\n",
      "|(min(Weight) + 10)|max((Weight + 10))|\n",
      "+------------------+------------------+\n",
      "|              1623|              5150|\n",
      "+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import min, max, lit\n",
    "df.select(min(col('Weight'))+lit(10), max(col('Weight')+lit(10))).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "KQ6Ul9HGCwC3"
   },
   "source": [
    "<a id='operations-on-date'></a>\n",
    "### Operations on Date"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "s1jmBN2qFHyk"
   },
   "source": [
    "> [PySpark follows SimpleDateFormat table of Java. Click here to view the docs.](https://docs.oracle.com/javase/7/docs/api/java/text/SimpleDateFormat.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "sCTeI_JvDCsH",
    "outputId": "e3f4a392-00d7-422d-945c-8038b3804876"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|                DOB|\n",
      "+-------------------+\n",
      "|2019-12-25 13:30:00|\n",
      "+-------------------+\n",
      "\n",
      "root\n",
      " |-- DOB: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import to_date, to_timestamp, lit\n",
    "df = spark.createDataFrame([('2019-12-25 13:30:00',)], ['DOB'])\n",
    "df.show()\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "ZH8ja1eHEW8x",
    "outputId": "2cef04c2-6c6f-494d-db9a-a4c17599b65c"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------------+--------------------------------------+\n",
      "|to_date(DOB, yyyy-MM-dd HH:mm:ss)|to_timestamp(DOB, yyyy-MM-dd HH:mm:ss)|\n",
      "+---------------------------------+--------------------------------------+\n",
      "|                       2019-12-25|                   2019-12-25 13:30:00|\n",
      "+---------------------------------+--------------------------------------+\n",
      "\n",
      "root\n",
      " |-- to_date(DOB, yyyy-MM-dd HH:mm:ss): date (nullable = true)\n",
      " |-- to_timestamp(DOB, yyyy-MM-dd HH:mm:ss): timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.createDataFrame([('2019-12-25 13:30:00',)], ['DOB'])\n",
    "df = df.select(to_date(col('DOB'),'yyyy-MM-dd HH:mm:ss'), to_timestamp(col('DOB'),'yyyy-MM-dd HH:mm:ss'))\n",
    "df.show()\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "7g9m_8PPErI1",
    "outputId": "854b3a71-bf83-4be8-c282-d94cc914c91f"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------+---------------------------------------+\n",
      "|to_date(DOB, dd/MMM/yyyy HH:mm:ss)|to_timestamp(DOB, dd/MMM/yyyy HH:mm:ss)|\n",
      "+----------------------------------+---------------------------------------+\n",
      "|                        2019-12-25|                    2019-12-25 13:30:00|\n",
      "+----------------------------------+---------------------------------------+\n",
      "\n",
      "root\n",
      " |-- to_date(DOB, dd/MMM/yyyy HH:mm:ss): date (nullable = true)\n",
      " |-- to_timestamp(DOB, dd/MMM/yyyy HH:mm:ss): timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.createDataFrame([('25/Dec/2019 13:30:00',)], ['DOB'])\n",
    "df = df.select(to_date(col('DOB'),'dd/MMM/yyyy HH:mm:ss'), to_timestamp(col('DOB'),'dd/MMM/yyyy HH:mm:ss'))\n",
    "df.show()\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "dIPQyQV7-Hz5"
   },
   "source": [
    "**What is 3 days earlier that the oldest date and 3 days later than the most recent date?**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "PUCEwQkZ-I7h",
    "outputId": "82621ccc-0113-4f4d-fb22-1551ab137893"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------+----------------------+\n",
      "|date_add(max(Date), 3)|date_sub(min(Date), 3)|\n",
      "+----------------------+----------------------+\n",
      "|            2021-04-02|            1989-12-29|\n",
      "+----------------------+----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import date_add, date_sub\n",
    "# create a dummy dataframe\n",
    "df = spark.createDataFrame([('1990-01-01',),('1995-01-03',),('2021-03-30',)], ['Date'])\n",
    "# find out the required dates\n",
    "df.select(date_add(max(col('Date')),3), date_sub(min(col('Date')),3)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "7OZElEvcGOD1"
   },
   "source": [
    "<a id='joins-in-pyspark'></a>\n",
    "## Joins in PySpark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "UJBC7r3JFyCL",
    "outputId": "65bafd3a-cc37-4a3a-e7ff-9578dd78dd4f"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "| id|car_name|\n",
      "+---+--------+\n",
      "|  1|   Car A|\n",
      "|  2|   Car B|\n",
      "|  3|   Car C|\n",
      "+---+--------+\n",
      "\n",
      "+---+---------+\n",
      "| id|car_price|\n",
      "+---+---------+\n",
      "|  1|     1000|\n",
      "|  2|     2000|\n",
      "|  3|     3000|\n",
      "+---+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create two dataframes\n",
    "cars_df = spark.createDataFrame([[1, 'Car A'],[2, 'Car B'],[3, 'Car C']], [\"id\", \"car_name\"])\n",
    "car_price_df = spark.createDataFrame([[1, 1000],[2, 2000],[3, 3000]], [\"id\", \"car_price\"])\n",
    "cars_df.show()\n",
    "car_price_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "U7Py4EYyKJTN",
    "outputId": "6bcd2bed-a2a1-4ef7-dbb5-20d27170ce05"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+---------+\n",
      "|id |car_name|car_price|\n",
      "+---+--------+---------+\n",
      "|1  |Car A   |1000     |\n",
      "|3  |Car C   |3000     |\n",
      "|2  |Car B   |2000     |\n",
      "+---+--------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Executing an inner join so we can see the id, name and price of each car in one row\n",
    "cars_df.join(car_price_df, cars_df.id == car_price_df.id, 'inner').select(cars_df['id'],cars_df['car_name'],car_price_df['car_price']).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "vj0mPaHU5i5n"
   },
   "source": [
    "As you can see, we have done an inner join between two dataframes. The following joins are supported by PySpark:\n",
    "1. inner (default)\n",
    "2. cross\n",
    "3. outer\n",
    "4. full\n",
    "5. full_outer\n",
    "6. left\n",
    "7. left_outer\n",
    "8. right\n",
    "9. right_outer\n",
    "10. left_semi\n",
    "11. left_anti"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "HNPhsx8P2tUH"
   },
   "source": [
    "<a id='spark-sql'></a>\n",
    "## Spark SQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "rHMvBBAh23cw"
   },
   "source": [
    "SQL has been around since the 1970s, and so one can imagine the number of people who made it their bread and butter. As big data came into popularity, the number of professionals with the technical knowledge to deal with it was in shortage. This led to the creation of Spark SQL. To quote the docs:<br>\n",
    ">Spark SQL is a Spark module for structured data processing. Unlike the basic Spark RDD API, the interfaces provided by Spark SQL provide Spark with more information about the structure of both the data and the computation being performed. Internally, Spark SQL uses this extra information to perform extra optimizations.\n",
    "\n",
    "Basically, what you need to know is that Spark SQL is used to execute SQL queries on big data. Spark SQL can also be used to read data from Hive tables and views. Let me explain Spark SQL with an example.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "g2DaK9-D7QkX",
    "outputId": "2f3156cd-7629-4021-fe8f-1f8b7980e650"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "|                 Car| MPG|Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|\n",
      "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "|Chevrolet Chevell...|18.0|        8|       307.0|     130.0| 3504.|        12.0|   70|    US|\n",
      "|   Buick Skylark 320|15.0|        8|       350.0|     165.0| 3693.|        11.5|   70|    US|\n",
      "|  Plymouth Satellite|18.0|        8|       318.0|     150.0| 3436.|        11.0|   70|    US|\n",
      "|       AMC Rebel SST|16.0|        8|       304.0|     150.0| 3433.|        12.0|   70|    US|\n",
      "|         Ford Torino|17.0|        8|       302.0|     140.0| 3449.|        10.5|   70|    US|\n",
      "+--------------------+----+---------+------------+----------+------+------------+-----+------+\n",
      "\n",
      "+-----------+\n",
      "|total_count|\n",
      "+-----------+\n",
      "|        406|\n",
      "+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Load data\n",
    "df = spark.read.csv('cars.csv', header=True, sep=\";\")\n",
    "# Register Temporary Table\n",
    "df.createOrReplaceTempView(\"temp\")\n",
    "# Select all data from temp table\n",
    "spark.sql(\"select * from temp limit 5\").show()\n",
    "# Select count of data in table\n",
    "spark.sql(\"select count(*) as total_count from temp\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "6i32WE8j_ec8"
   },
   "source": [
    "As you can see, we registered the dataframe as temporary table and then ran basic SQL queries on it. How amazing is that?!<br>\n",
    "If you are a person who is more comfortable with SQL, then this feature is truly a blessing for you! But this raises a question: \n",
    "> *Should I just keep using Spark SQL all the time?*\n",
    "\n",
    "And the answer is, _**it depends**_.<br>\n",
    "So basically, the different functions acts in differnet ways, and depending upon the type of action you are trying to do, the speed at which it completes execution also differs. But as time progress, this feature is getting better and better, so hopefully the difference should be a small margin. There are plenty of analysis done on this, but nothing has a definite answer yet. You can read this [comparative study done by horton works](https://community.cloudera.com/t5/Community-Articles/Spark-RDDs-vs-DataFrames-vs-SparkSQL/ta-p/246547) or the answer to this [stackoverflow question](https://stackoverflow.com/questions/45430816/writing-sql-vs-using-dataframe-apis-in-spark-sql) if you are still curious about it."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "x62BiCgBMOtq"
   },
   "source": [
    "<a id='rdd'></a>\n",
    "## RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "VGXK6uEuUKRh"
   },
   "source": [
    "> With map, you define a function and then apply it record by record. Flatmap returns a new RDD by first applying a function to all of the elements in RDDs and then flattening the result. Filter, returns a new RDD. Meaning only the elements that satisfy a condition. With reduce, we are taking neighboring elements and producing a single combined result.\n",
    "For example, let's say you have a set of numbers. You can reduce this to its sum by providing a function that takes as input two values and reduces them to one. \n",
    "\n",
    "Some of the reasons you would use a dataframe over RDD are:\n",
    "1.   It's ability to represnt data as rows and columns. But this also means it can only hold structred and semi-structured data.\n",
    "2.   It allows processing data in different formats (AVRO, CSV, JSON, and storage system HDFS, HIVE tables, MySQL).\n",
    "3. It's superior job Optimization capability.\n",
    "4. DataFrame API is very easy to use.\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "0_WvAgyvR7m6",
    "outputId": "aca36808-b04a-4a45-d628-aeb6b5bab927"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Car;MPG;Cylinders;Displacement;Horsepower;Weight;Acceleration;Model;Origin\n",
      "Chevrolet Chevelle Malibu;18.0;8;307.0;130.0;3504.;12.0;70;US\n"
     ]
    }
   ],
   "source": [
    "cars = spark.sparkContext.textFile('cars.csv')\n",
    "print(cars.first())\n",
    "cars_header = cars.first()\n",
    "cars_rest = cars.filter(lambda line: line!=cars_header)\n",
    "print(cars_rest.first())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "P65eAFO3Mkdd"
   },
   "source": [
    "**How many cars are there in our csv data?**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Vi03EU0CMSmO",
    "outputId": "2215311f-b291-4659-b2ac-4f3d4491531f"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "406"
      ]
     },
     "execution_count": 61,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cars_rest.map(lambda line: line.split(\";\")).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "3c4bci70MnlQ"
   },
   "source": [
    "**Display the Car name, MPG, Cylinders, Weight and Origin for the cars Originating in Europe**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "fWFpo_WxMnvm",
    "outputId": "db9ccb02-d380-4b62-8ff8-679ed5e06e11"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Citroen DS-21 Pallas', '0', '4', '3090.', 'Europe'),\n",
       " ('Volkswagen 1131 Deluxe Sedan', '26.0', '4', '1835.', 'Europe'),\n",
       " ('Peugeot 504', '25.0', '4', '2672.', 'Europe'),\n",
       " ('Audi 100 LS', '24.0', '4', '2430.', 'Europe'),\n",
       " ('Saab 99e', '25.0', '4', '2375.', 'Europe'),\n",
       " ('BMW 2002', '26.0', '4', '2234.', 'Europe'),\n",
       " ('Volkswagen Super Beetle 117', '0', '4', '1978.', 'Europe'),\n",
       " ('Opel 1900', '28.0', '4', '2123.', 'Europe'),\n",
       " ('Peugeot 304', '30.0', '4', '2074.', 'Europe'),\n",
       " ('Fiat 124B', '30.0', '4', '2065.', 'Europe'),\n",
       " ('Volkswagen Model 111', '27.0', '4', '1834.', 'Europe'),\n",
       " ('Volkswagen Type 3', '23.0', '4', '2254.', 'Europe'),\n",
       " ('Volvo 145e (sw)', '18.0', '4', '2933.', 'Europe'),\n",
       " ('Volkswagen 411 (sw)', '22.0', '4', '2511.', 'Europe'),\n",
       " ('Peugeot 504 (sw)', '21.0', '4', '2979.', 'Europe'),\n",
       " ('Renault 12 (sw)', '26.0', '4', '2189.', 'Europe'),\n",
       " ('Volkswagen Super Beetle', '26.0', '4', '1950.', 'Europe'),\n",
       " ('Fiat 124 Sport Coupe', '26.0', '4', '2265.', 'Europe'),\n",
       " ('Fiat 128', '29.0', '4', '1867.', 'Europe'),\n",
       " ('Opel Manta', '24.0', '4', '2158.', 'Europe'),\n",
       " ('Audi 100LS', '20.0', '4', '2582.', 'Europe'),\n",
       " ('Volvo 144ea', '19.0', '4', '2868.', 'Europe'),\n",
       " ('Saab 99le', '24.0', '4', '2660.', 'Europe'),\n",
       " ('Audi Fox', '29.0', '4', '2219.', 'Europe'),\n",
       " ('Volkswagen Dasher', '26.0', '4', '1963.', 'Europe'),\n",
       " ('Opel Manta', '26.0', '4', '2300.', 'Europe'),\n",
       " ('Fiat 128', '24.0', '4', '2108.', 'Europe'),\n",
       " ('Fiat 124 TC', '26.0', '4', '2246.', 'Europe'),\n",
       " ('Fiat x1.9', '31.0', '4', '2000.', 'Europe'),\n",
       " ('Volkswagen Dasher', '25.0', '4', '2223.', 'Europe'),\n",
       " ('Volkswagen Rabbit', '29.0', '4', '1937.', 'Europe'),\n",
       " ('Audi 100LS', '23.0', '4', '2694.', 'Europe'),\n",
       " ('Peugeot 504', '23.0', '4', '2957.', 'Europe'),\n",
       " ('Volvo 244DL', '22.0', '4', '2945.', 'Europe'),\n",
       " ('Saab 99LE', '25.0', '4', '2671.', 'Europe'),\n",
       " ('Fiat 131', '28.0', '4', '2464.', 'Europe'),\n",
       " ('Opel 1900', '25.0', '4', '2220.', 'Europe'),\n",
       " ('Renault 12tl', '27.0', '4', '2202.', 'Europe'),\n",
       " ('Volkswagen Rabbit', '29.0', '4', '1937.', 'Europe'),\n",
       " ('Volkswagen Rabbit', '29.5', '4', '1825.', 'Europe'),\n",
       " ('Volvo 245', '20.0', '4', '3150.', 'Europe'),\n",
       " ('Peugeot 504', '19.0', '4', '3270.', 'Europe'),\n",
       " ('Mercedes-Benz 280s', '16.5', '6', '3820.', 'Europe'),\n",
       " ('Renault 5 GTL', '36.0', '4', '1825.', 'Europe'),\n",
       " ('Volkswagen Rabbit Custom', '29.0', '4', '1940.', 'Europe'),\n",
       " ('Volkswagen Dasher', '30.5', '4', '2190.', 'Europe'),\n",
       " ('BMW 320i', '21.5', '4', '2600.', 'Europe'),\n",
       " ('Volkswagen Rabbit Custom Diesel', '43.1', '4', '1985.', 'Europe'),\n",
       " ('Audi 5000', '20.3', '5', '2830.', 'Europe'),\n",
       " ('Volvo 264gl', '17.0', '6', '3140.', 'Europe'),\n",
       " ('Saab 99gle', '21.6', '4', '2795.', 'Europe'),\n",
       " ('Peugeot 604sl', '16.2', '6', '3410.', 'Europe'),\n",
       " ('Volkswagen Scirocco', '31.5', '4', '1990.', 'Europe'),\n",
       " ('Volkswagen Rabbit Custom', '31.9', '4', '1925.', 'Europe'),\n",
       " ('Mercedes Benz 300d', '25.4', '5', '3530.', 'Europe'),\n",
       " ('Peugeot 504', '27.2', '4', '3190.', 'Europe'),\n",
       " ('Fiat Strada Custom', '37.3', '4', '2130.', 'Europe'),\n",
       " ('Volkswagen Rabbit', '41.5', '4', '2144.', 'Europe'),\n",
       " ('Audi 4000', '34.3', '4', '2188.', 'Europe'),\n",
       " ('Volkswagen Rabbit C (Diesel)', '44.3', '4', '2085.', 'Europe'),\n",
       " ('Volkswagen Dasher (diesel)', '43.4', '4', '2335.', 'Europe'),\n",
       " ('Audi 5000s (diesel)', '36.4', '5', '2950.', 'Europe'),\n",
       " ('Mercedes-Benz 240d', '30.0', '4', '3250.', 'Europe'),\n",
       " ('Renault Lecar Deluxe', '40.9', '4', '1835.', 'Europe'),\n",
       " ('Volkswagen Rabbit', '29.8', '4', '1845.', 'Europe'),\n",
       " ('Triumph TR7 Coupe', '35.0', '4', '2500.', 'Europe'),\n",
       " ('Volkswagen Jetta', '33.0', '4', '2190.', 'Europe'),\n",
       " ('Renault 18i', '34.5', '4', '2320.', 'Europe'),\n",
       " ('Peugeot 505s Turbo Diesel', '28.1', '4', '3230.', 'Europe'),\n",
       " ('Saab 900s', '0', '4', '2800.', 'Europe'),\n",
       " ('Volvo Diesel', '30.7', '6', '3160.', 'Europe'),\n",
       " ('Volkswagen Rabbit l', '36.0', '4', '1980.', 'Europe'),\n",
       " ('Volkswagen Pickup', '44.0', '4', '2130.', 'Europe')]"
      ]
     },
     "execution_count": 62,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Car name is column  0\n",
    "(cars_rest.filter(lambda line: line.split(\";\")[8]=='Europe').\n",
    " map(lambda line: (line.split(\";\")[0],\n",
    "    line.split(\";\")[1],\n",
    "    line.split(\";\")[2],\n",
    "    line.split(\";\")[5],\n",
    "    line.split(\";\")[8])).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ZYmb5FscMph3"
   },
   "source": [
    "**Display the Car name, MPG, Cylinders, Weight and Origin for the cars Originating in either Europe or Japan**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "6ZcRIX3mMquF",
    "outputId": "8d803b6d-02f3-49dd-ef89-6178292d0217"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Citroen DS-21 Pallas', '0', '4', '3090.', 'Europe'),\n",
       " ('Toyota Corolla Mark ii', '24.0', '4', '2372.', 'Japan'),\n",
       " ('Datsun PL510', '27.0', '4', '2130.', 'Japan'),\n",
       " ('Volkswagen 1131 Deluxe Sedan', '26.0', '4', '1835.', 'Europe'),\n",
       " ('Peugeot 504', '25.0', '4', '2672.', 'Europe'),\n",
       " ('Audi 100 LS', '24.0', '4', '2430.', 'Europe'),\n",
       " ('Saab 99e', '25.0', '4', '2375.', 'Europe'),\n",
       " ('BMW 2002', '26.0', '4', '2234.', 'Europe'),\n",
       " ('Datsun PL510', '27.0', '4', '2130.', 'Japan'),\n",
       " ('Toyota Corolla', '25.0', '4', '2228.', 'Japan'),\n",
       " ('Volkswagen Super Beetle 117', '0', '4', '1978.', 'Europe'),\n",
       " ('Opel 1900', '28.0', '4', '2123.', 'Europe'),\n",
       " ('Peugeot 304', '30.0', '4', '2074.', 'Europe'),\n",
       " ('Fiat 124B', '30.0', '4', '2065.', 'Europe'),\n",
       " ('Toyota Corolla 1200', '31.0', '4', '1773.', 'Japan'),\n",
       " ('Datsun 1200', '35.0', '4', '1613.', 'Japan'),\n",
       " ('Volkswagen Model 111', '27.0', '4', '1834.', 'Europe'),\n",
       " ('Toyota Corolla Hardtop', '24.0', '4', '2278.', 'Japan'),\n",
       " ('Volkswagen Type 3', '23.0', '4', '2254.', 'Europe'),\n",
       " ('Mazda RX2 Coupe', '19.0', '3', '2330.', 'Japan'),\n",
       " ('Volvo 145e (sw)', '18.0', '4', '2933.', 'Europe'),\n",
       " ('Volkswagen 411 (sw)', '22.0', '4', '2511.', 'Europe'),\n",
       " ('Peugeot 504 (sw)', '21.0', '4', '2979.', 'Europe'),\n",
       " ('Renault 12 (sw)', '26.0', '4', '2189.', 'Europe'),\n",
       " ('Datsun 510 (sw)', '28.0', '4', '2288.', 'Japan'),\n",
       " ('Toyota Corolla Mark II (sw)', '23.0', '4', '2506.', 'Japan'),\n",
       " ('Toyota Corolla 1600 (sw)', '27.0', '4', '2100.', 'Japan'),\n",
       " ('Volkswagen Super Beetle', '26.0', '4', '1950.', 'Europe'),\n",
       " ('Toyota Camry', '20.0', '4', '2279.', 'Japan'),\n",
       " ('Datsun 610', '22.0', '4', '2379.', 'Japan'),\n",
       " ('Mazda RX3', '18.0', '3', '2124.', 'Japan'),\n",
       " ('Fiat 124 Sport Coupe', '26.0', '4', '2265.', 'Europe'),\n",
       " ('Fiat 128', '29.0', '4', '1867.', 'Europe'),\n",
       " ('Opel Manta', '24.0', '4', '2158.', 'Europe'),\n",
       " ('Audi 100LS', '20.0', '4', '2582.', 'Europe'),\n",
       " ('Volvo 144ea', '19.0', '4', '2868.', 'Europe'),\n",
       " ('Saab 99le', '24.0', '4', '2660.', 'Europe'),\n",
       " ('Toyota Mark II', '20.0', '6', '2807.', 'Japan'),\n",
       " ('Datsun B210', '31.0', '4', '1950.', 'Japan'),\n",
       " ('Toyota Corolla 1200', '32.0', '4', '1836.', 'Japan'),\n",
       " ('Audi Fox', '29.0', '4', '2219.', 'Europe'),\n",
       " ('Volkswagen Dasher', '26.0', '4', '1963.', 'Europe'),\n",
       " ('Opel Manta', '26.0', '4', '2300.', 'Europe'),\n",
       " ('Toyota Corolla', '31.0', '4', '1649.', 'Japan'),\n",
       " ('Datsun 710', '32.0', '4', '2003.', 'Japan'),\n",
       " ('Fiat 128', '24.0', '4', '2108.', 'Europe'),\n",
       " ('Fiat 124 TC', '26.0', '4', '2246.', 'Europe'),\n",
       " ('Honda Civic', '24.0', '4', '2489.', 'Japan'),\n",
       " ('Subaru', '26.0', '4', '2391.', 'Japan'),\n",
       " ('Fiat x1.9', '31.0', '4', '2000.', 'Europe'),\n",
       " ('Toyota Corolla', '29.0', '4', '2171.', 'Japan'),\n",
       " ('Toyota Corolla', '24.0', '4', '2702.', 'Japan'),\n",
       " ('Volkswagen Dasher', '25.0', '4', '2223.', 'Europe'),\n",
       " ('Datsun 710', '24.0', '4', '2545.', 'Japan'),\n",
       " ('Volkswagen Rabbit', '29.0', '4', '1937.', 'Europe'),\n",
       " ('Audi 100LS', '23.0', '4', '2694.', 'Europe'),\n",
       " ('Peugeot 504', '23.0', '4', '2957.', 'Europe'),\n",
       " ('Volvo 244DL', '22.0', '4', '2945.', 'Europe'),\n",
       " ('Saab 99LE', '25.0', '4', '2671.', 'Europe'),\n",
       " ('Honda Civic CVCC', '33.0', '4', '1795.', 'Japan'),\n",
       " ('Fiat 131', '28.0', '4', '2464.', 'Europe'),\n",
       " ('Opel 1900', '25.0', '4', '2220.', 'Europe'),\n",
       " ('Renault 12tl', '27.0', '4', '2202.', 'Europe'),\n",
       " ('Volkswagen Rabbit', '29.0', '4', '1937.', 'Europe'),\n",
       " ('Honda Civic', '33.0', '4', '1795.', 'Japan'),\n",
       " ('Volkswagen Rabbit', '29.5', '4', '1825.', 'Europe'),\n",
       " ('Datsun B-210', '32.0', '4', '1990.', 'Japan'),\n",
       " ('Toyota Corolla', '28.0', '4', '2155.', 'Japan'),\n",
       " ('Volvo 245', '20.0', '4', '3150.', 'Europe'),\n",
       " ('Peugeot 504', '19.0', '4', '3270.', 'Europe'),\n",
       " ('Toyota Mark II', '19.0', '6', '2930.', 'Japan'),\n",
       " ('Mercedes-Benz 280s', '16.5', '6', '3820.', 'Europe'),\n",
       " ('Honda Accord CVCC', '31.5', '4', '2045.', 'Japan'),\n",
       " ('Renault 5 GTL', '36.0', '4', '1825.', 'Europe'),\n",
       " ('Datsun F-10 Hatchback', '33.5', '4', '1945.', 'Japan'),\n",
       " ('Volkswagen Rabbit Custom', '29.0', '4', '1940.', 'Europe'),\n",
       " ('Toyota Corolla Liftback', '26.0', '4', '2265.', 'Japan'),\n",
       " ('Subaru DL', '30.0', '4', '1985.', 'Japan'),\n",
       " ('Volkswagen Dasher', '30.5', '4', '2190.', 'Europe'),\n",
       " ('Datsun 810', '22.0', '6', '2815.', 'Japan'),\n",
       " ('BMW 320i', '21.5', '4', '2600.', 'Europe'),\n",
       " ('Mazda RX-4', '21.5', '3', '2720.', 'Japan'),\n",
       " ('Volkswagen Rabbit Custom Diesel', '43.1', '4', '1985.', 'Europe'),\n",
       " ('Mazda GLC Deluxe', '32.8', '4', '1985.', 'Japan'),\n",
       " ('Datsun B210 GX', '39.4', '4', '2070.', 'Japan'),\n",
       " ('Honda Civic CVCC', '36.1', '4', '1800.', 'Japan'),\n",
       " ('Toyota Corolla', '27.5', '4', '2560.', 'Japan'),\n",
       " ('Datsun 510', '27.2', '4', '2300.', 'Japan'),\n",
       " ('Toyota Celica GT Liftback', '21.1', '4', '2515.', 'Japan'),\n",
       " ('Datsun 200-SX', '23.9', '4', '2405.', 'Japan'),\n",
       " ('Audi 5000', '20.3', '5', '2830.', 'Europe'),\n",
       " ('Volvo 264gl', '17.0', '6', '3140.', 'Europe'),\n",
       " ('Saab 99gle', '21.6', '4', '2795.', 'Europe'),\n",
       " ('Peugeot 604sl', '16.2', '6', '3410.', 'Europe'),\n",
       " ('Volkswagen Scirocco', '31.5', '4', '1990.', 'Europe'),\n",
       " ('Honda Accord LX', '29.5', '4', '2135.', 'Japan'),\n",
       " ('Volkswagen Rabbit Custom', '31.9', '4', '1925.', 'Europe'),\n",
       " ('Mazda GLC Deluxe', '34.1', '4', '1975.', 'Japan'),\n",
       " ('Mercedes Benz 300d', '25.4', '5', '3530.', 'Europe'),\n",
       " ('Peugeot 504', '27.2', '4', '3190.', 'Europe'),\n",
       " ('Datsun 210', '31.8', '4', '2020.', 'Japan'),\n",
       " ('Fiat Strada Custom', '37.3', '4', '2130.', 'Europe'),\n",
       " ('Volkswagen Rabbit', '41.5', '4', '2144.', 'Europe'),\n",
       " ('Toyota Corolla Tercel', '38.1', '4', '1968.', 'Japan'),\n",
       " ('Datsun 310', '37.2', '4', '2019.', 'Japan'),\n",
       " ('Audi 4000', '34.3', '4', '2188.', 'Europe'),\n",
       " ('Toyota Corolla Liftback', '29.8', '4', '2711.', 'Japan'),\n",
       " ('Mazda 626', '31.3', '4', '2542.', 'Japan'),\n",
       " ('Datsun 510 Hatchback', '37.0', '4', '2434.', 'Japan'),\n",
       " ('Toyota Corolla', '32.2', '4', '2265.', 'Japan'),\n",
       " ('Mazda GLC', '46.6', '4', '2110.', 'Japan'),\n",
       " ('Datsun 210', '40.8', '4', '2110.', 'Japan'),\n",
       " ('Volkswagen Rabbit C (Diesel)', '44.3', '4', '2085.', 'Europe'),\n",
       " ('Volkswagen Dasher (diesel)', '43.4', '4', '2335.', 'Europe'),\n",
       " ('Audi 5000s (diesel)', '36.4', '5', '2950.', 'Europe'),\n",
       " ('Mercedes-Benz 240d', '30.0', '4', '3250.', 'Europe'),\n",
       " ('Honda Civic 1500 gl', '44.6', '4', '1850.', 'Japan'),\n",
       " ('Renault Lecar Deluxe', '40.9', '4', '1835.', 'Europe'),\n",
       " ('Subaru DL', '33.8', '4', '2145.', 'Japan'),\n",
       " ('Volkswagen Rabbit', '29.8', '4', '1845.', 'Europe'),\n",
       " ('Datsun 280-ZX', '32.7', '6', '2910.', 'Japan'),\n",
       " ('Mazda RX-7 GS', '23.7', '3', '2420.', 'Japan'),\n",
       " ('Triumph TR7 Coupe', '35.0', '4', '2500.', 'Europe'),\n",
       " ('Honda Accord', '32.4', '4', '2290.', 'Japan'),\n",
       " ('Toyota Starlet', '39.1', '4', '1755.', 'Japan'),\n",
       " ('Honda Civic 1300', '35.1', '4', '1760.', 'Japan'),\n",
       " ('Subaru', '32.3', '4', '2065.', 'Japan'),\n",
       " ('Datsun 210 MPG', '37.0', '4', '1975.', 'Japan'),\n",
       " ('Toyota Tercel', '37.7', '4', '2050.', 'Japan'),\n",
       " ('Mazda GLC 4', '34.1', '4', '1985.', 'Japan'),\n",
       " ('Volkswagen Jetta', '33.0', '4', '2190.', 'Europe'),\n",
       " ('Renault 18i', '34.5', '4', '2320.', 'Europe'),\n",
       " ('Honda Prelude', '33.7', '4', '2210.', 'Japan'),\n",
       " ('Toyota Corolla', '32.4', '4', '2350.', 'Japan'),\n",
       " ('Datsun 200SX', '32.9', '4', '2615.', 'Japan'),\n",
       " ('Mazda 626', '31.6', '4', '2635.', 'Japan'),\n",
       " ('Peugeot 505s Turbo Diesel', '28.1', '4', '3230.', 'Europe'),\n",
       " ('Saab 900s', '0', '4', '2800.', 'Europe'),\n",
       " ('Volvo Diesel', '30.7', '6', '3160.', 'Europe'),\n",
       " ('Toyota Cressida', '25.4', '6', '2900.', 'Japan'),\n",
       " ('Datsun 810 Maxima', '24.2', '6', '2930.', 'Japan'),\n",
       " ('Volkswagen Rabbit l', '36.0', '4', '1980.', 'Europe'),\n",
       " ('Mazda GLC Custom l', '37.0', '4', '2025.', 'Japan'),\n",
       " ('Mazda GLC Custom', '31.0', '4', '1970.', 'Japan'),\n",
       " ('Nissan Stanza XE', '36.0', '4', '2160.', 'Japan'),\n",
       " ('Honda Accord', '36.0', '4', '2205.', 'Japan'),\n",
       " ('Toyota Corolla', '34.0', '4', '2245', 'Japan'),\n",
       " ('Honda Civic', '38.0', '4', '1965.', 'Japan'),\n",
       " ('Honda Civic (auto)', '32.0', '4', '1965.', 'Japan'),\n",
       " ('Datsun 310 GX', '38.0', '4', '1995.', 'Japan'),\n",
       " ('Toyota Celica GT', '32.0', '4', '2665.', 'Japan'),\n",
       " ('Volkswagen Pickup', '44.0', '4', '2130.', 'Europe')]"
      ]
     },
     "execution_count": 63,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Car name is column  0\n",
    "(cars_rest.filter(lambda line: line.split(\";\")[8] in ['Europe','Japan']).\n",
    " map(lambda line: (line.split(\";\")[0],\n",
    "    line.split(\";\")[1],\n",
    "    line.split(\";\")[2],\n",
    "    line.split(\";\")[5],\n",
    "    line.split(\";\")[8])).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "3wn2zXe7TbI3"
   },
   "source": [
    "<a id='user-defined-functions-udf'></a>\n",
    "## User-Defined Functions (UDF)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "w0YWspcTRrin"
   },
   "source": [
    "PySpark User-Defined Functions (UDFs) help you convert your python code into a scalable version of itself. It comes in handy more than you can imagine, but beware, as the performance is less when you compare it with pyspark functions. You can view examples of how UDF works [here](https://docs.databricks.com/spark/latest/spark-sql/udf-python.html). What I will give in this section is some theory on how it works, and why it is slower.\n",
    "\n",
    "When you try to run a UDF in PySpark, each executor creates a python process. Data will be serialised and deserialised between each executor and python. This leads to lots of performance impact and overhead on spark jobs, making it less efficent than using spark dataframes. Apart from this, sometimes you might have memory issues while using UDFs. The Python worker consumes huge off-heap memory and so it often leads to memoryOverhead, thereby failing your job. Keeping these in mind, I wouldn't recommend using them, but at the end of the day, your choice."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "yv7ODDTQRwVt"
   },
   "source": [
    "<a id='common-questions'></a>\n",
    "# Common Questions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "6z9gkoE2R1m1"
   },
   "source": [
    "<a id='recommended-ide'></a>\n",
    "## Recommended IDE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "BpMTbRggR5Z8"
   },
   "source": [
    "I personally prefer [PyCharm](https://www.jetbrains.com/pycharm/) while coding in Python/PySpark. It's based on IntelliJ IDEA so it has a lot of features! And the main advantage I have felt is the ease of installing PySpark and other packages. You can customize it with themes and plugins, and it lets you enhance productivity while coding by providing some features like suggestions, local VCS etc."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "VZ1bYvF8R8Dc"
   },
   "source": [
    "<a id='submitting-a-spark-job'></a>\n",
    "## Submitting a Spark Job"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "3EQWnq23SCbE"
   },
   "source": [
    "The python syntax for running jobs is: `python <file_name>.py <arg1> <arg2> ...`\n",
    "<br>But when you submit a spark job you have to use spark-submit to run the application.\n",
    "\n",
    "Here is a simple example of a spark-submit command:\n",
    "`spark-submit filename.py --named_argument 'arguemnt value'`<br>\n",
    "Here, named_argument is an argument that you are reading from inside your script.\n",
    "\n",
    "There are other options you can pass in the command, like:<br>\n",
    "`--py-files` which helps you pass a python file to read in your file,<br>\n",
    "`--files` which helps pass other files like txt or config,<br>\n",
    "`--deploy-mode` which tells wether to deploy your worker node on cluster or locally <br>\n",
    "`--conf` which helps pass different configurations, like memoryOverhead, dynamicAllocation etc.\n",
    "\n",
    "There is an [entire page](https://spark.apache.org/docs/latest/submitting-applications.html) in spark documentation dedicated to this. I highly recommend you go through it once."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "oVwGYAZZiyGV"
   },
   "source": [
    "<a id='creating-dataframes'></a>\n",
    "## Creating Dataframes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "TvndhPjoi0er"
   },
   "source": [
    "When getting started with dataframes, the most common question is: *'How do I create a dataframe?'* <br> \n",
    "Below, you can see how to create three kinds of dataframes:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "QXmRD3hHlM-f"
   },
   "source": [
    "### Create a totally empty dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "ktkb6s-kjtgG",
    "outputId": "08e2c46d-2a17-44c8-e954-37704b287f48"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "++\n",
      "||\n",
      "++\n",
      "++\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import StructType\n",
    "sc = spark.sparkContext\n",
    "#Create empty df\n",
    "schema = StructType([])\n",
    "empty = spark.createDataFrame(sc.emptyRDD(), schema)\n",
    "empty.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "mg5K3nz_lSDe"
   },
   "source": [
    "### Create an empty dataframe with header"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "9raf4CkRjuTr",
    "outputId": "afcce8fb-9136-4ec5-91b6-b6924dcea0e1"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "|name|\n",
      "+----+\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import StructType, StructField\n",
    "#Create empty df with header\n",
    "schema_header = StructType([StructField(\"name\", StringType(), True)])\n",
    "empty_with_header = spark.createDataFrame(sc.emptyRDD(), schema_header)\n",
    "empty_with_header.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Y1ZNOx7ilUnd"
   },
   "source": [
    "### Create a dataframe with header and data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "TvzyL46QkJBl",
    "outputId": "80bd5c29-cd11-4226-be60-a0db525651f6"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+---+\n",
      "| name|age|\n",
      "+-----+---+\n",
      "|Alice| 13|\n",
      "|Jacob| 24|\n",
      "|Betty|135|\n",
      "+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import Row\n",
    "mylist = [\n",
    "  {\"name\":'Alice',\"age\":13},\n",
    "  {\"name\":'Jacob',\"age\":24},\n",
    "  {\"name\":'Betty',\"age\":135},\n",
    "]\n",
    "spark.createDataFrame(Row(**x) for x in mylist).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "VnRMckA5nLoJ",
    "outputId": "f362d23c-5343-4601-d669-0e26940c043a"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+---+\n",
      "| name|age|\n",
      "+-----+---+\n",
      "|Alice| 13|\n",
      "|Jacob| 24|\n",
      "|Betty|135|\n",
      "+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# You can achieve the same using this - note that we are using spark context here, not a spark session\n",
    "from pyspark.sql import Row\n",
    "df = sc.parallelize([\n",
    "        Row(name='Alice', age=13),\n",
    "        Row(name='Jacob', age=24),\n",
    "        Row(name='Betty', age=135)]).toDF()\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "f3crkAQVlxKp"
   },
   "source": [
    "<a id='drop-duplicates'></a>\n",
    "## Drop Duplicates"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "4IHrYEwHmBcc"
   },
   "source": [
    "As mentioned earlier, there are two easy to remove duplicates from a dataframe. We have already seen the usage of distinct under <a href=\"#get-distinct-rows\">Get Distinct Rows</a>  section. \n",
    "I will expalin how to use the `dropDuplicates()` function to achieve the same. \n",
    "\n",
    "> `drop_duplicates()` is an alias for `dropDuplicates()`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "wOuHRAPJmWen",
    "outputId": "08dcf7e7-91b3-4e42-b418-e29cafd1d1f6"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+---+------+\n",
      "| name|age|height|\n",
      "+-----+---+------+\n",
      "|Alice|  5|    80|\n",
      "|Jacob| 24|    80|\n",
      "+-----+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import Row\n",
    "from pyspark.sql import Row\n",
    "mylist = [\n",
    "  {\"name\":'Alice',\"age\":5,\"height\":80},\n",
    "  {\"name\":'Jacob',\"age\":24,\"height\":80},\n",
    "  {\"name\":'Alice',\"age\":5,\"height\":80}\n",
    "]\n",
    "df = spark.createDataFrame(Row(**x) for x in mylist)\n",
    "df.dropDuplicates().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "zMv7A-2Hnmjh"
   },
   "source": [
    "`dropDuplicates()` can also take in an optional parameter called *subset* which helps specify the columns on which the duplicate check needs to be done on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "SHnFylV1n8to",
    "outputId": "04d2b333-5770-4287-bb48-775bbc08aad2"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+---+------+\n",
      "| name|age|height|\n",
      "+-----+---+------+\n",
      "|Alice|  5|    80|\n",
      "+-----+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.dropDuplicates(subset=['height']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "bAS4DKxjqI7H"
   },
   "source": [
    "<a id='fine-tuning-a-pyspark-job'></a>\n",
    "## Fine Tuning a Spark Job"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "2d9pLlDl76dM"
   },
   "source": [
    "Before we begin, please note that this entire section is written purely based on experience. It might differ with use cases, but it will help you get a better understanding of what you should be looking for, or act as a guidance to achieve your aim.\n",
    "\n",
    ">Spark Performance Tuning refers to the process of adjusting settings to record for memory, cores, and instances used by the system. This process guarantees that the Spark has a flawless performance and also prevents bottlenecking of resources in Spark.\n",
    "\n",
    "Considering you are using Amazon EMR to execute your spark jobs, there are three aspects you need to take care of:\n",
    "1. EMR Sizing\n",
    "2. Spark Configurations\n",
    "3. Job Tuning\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "vIbXZT29JxmG"
   },
   "source": [
    "<a id='emr-sizing'></a>\n",
    "### EMR Sizing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "2Rv2SM_-KA8W"
   },
   "source": [
    "Sizing your EMR is extremely important, as this affects the efficency of your spark jobs. Apart from the cost factor, the maximum number of nodes and memory your job can use will be decided by this. If you spin up a EMR with high specifications, that obviously means you are paying more for it, so we should ideally utilize it to the max. These are the guidelines that I follow to make sure the EMR is rightly sized:\n",
    "\n",
    "1. Size of the input data (include all the input data) on the disk.\n",
    "2. Whether the jobs have transformations or just a straight pass through.<br> Assess the joins and the complex joins involved.\n",
    "3. Size of the output data on the disk.\n",
    "\n",
    "Look at the above criteria against the memory you need to process, and the disk space you would need. Start with a small configuration, and keep adding nodes to arrive at an optimal configuration. In case you are wondering about the *Execution time vs EMR configuration* factor, please understand that it is okay for a job to run longer, rather than adding more resources to the cluster. For example, it is okay to run a job for 40 mins job on a 5 node cluster, rather than running a job in 10 mins on a 15 node cluster.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "pmeFEgv6QTST"
   },
   "source": [
    "Another thing you need to know about EMRs, are the different kinds of EC2 instance types provided by Amazon. I will briefly talk about them, but I strongly recommend you to read more about it from the [official documentation](https://aws.amazon.com/ec2/instance-types/). There are 5 types of instance classes. Based on the job you want to run, you can decide which one to use:\n",
    "\n",
    ">Instance Class | Description\n",
    ">--- | ---\n",
    ">General purpose | Balance of compute, memory and networking resources\n",
    ">Compute optimized | Ideal for compute bound applications that benefit from high performance processors\n",
    ">Memory optimized | Designed to deliver fast performance for workloads that process large data sets in memory\n",
    ">Storage optimized | For workloads that require high, sequential read and write access to very large data sets on local storage\n",
    ">GPU instances | Use hardware accelerators, or co-processors, to perform high demanding functions, more efficiently than is possible in software running on CPUs\n",
    "\n",
    "The configuration (memory, storage, cpu, network performance) will differ based on the instance class you choose.<br>\n",
    "To help make life easier, here is what I do when I get into a predicament about which one to go with: <br>\n",
    " 1. Visit [ec2instances](https://www.ec2instances.info/)\n",
    " 2. Choose the EC2 instances in question \n",
    " 3. Click on compare selected\n",
    "\n",
    "This will easily help you undesrstand what you are getting into, and thereby help you make the best choice! The site was built by [Garret Heaton](https://github.com/powdahound)(founder of Swoot), and has helped me countless number of times to make an informed decision."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "snACMwZug5Yn"
   },
   "source": [
    "<a id='spark-configurations'></a>\n",
    "### Spark Configurations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "uFJNpK06hnpo"
   },
   "source": [
    "There are a ton of [configurations](https://spark.apache.org/docs/latest/configuration.html) that you can tweak when it comes to Spark. Here, I will be noting down some of the configurations which I use, which have worked well for me. Alright! let's get into it!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "dABRu9eokZxw"
   },
   "source": [
    "#### Job Scheduling"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "8fRFs6atkdxS"
   },
   "source": [
    "When you submit your job in a cluster, it will be given to Spark Schedulers, which is responsible for materializing a logical plan for your job. There are two types of [job scheduling](https://spark.apache.org/docs/latest/job-scheduling.html): \n",
    "1. FIFO<br>\n",
    "By default, Spark’s scheduler runs jobs in FIFO fashion. Each job is divided into stages (e.g. map and reduce phases), and the first job gets priority on all available resources while its stages have tasks to launch, then the second job gets priority, etc. If the jobs at the head of the queue don’t need to use the whole cluster, later jobs can start to run right away, but if the jobs at the head of the queue are large, then later jobs may be delayed significantly. \n",
    "2. FAIR<br>\n",
    "The fair scheduler supports grouping jobs into pools and setting different scheduling options (e.g. weight) for each pool. This can be useful to create a high-priority pool for more important jobs, for example, or to group the jobs of each user together and give users equal shares regardless of how many concurrent jobs they have instead of giving jobs equal shares. This approach is modeled after the Hadoop Fair Scheduler.\n",
    "\n",
    "> I personally prefer using the FAIR mode, and this can be set by adding `.config(\"spark.scheduler.mode\", \"FAIR\")` when you create your SparkSession.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "xrgbAiZHnq_U"
   },
   "source": [
    "#### Serializer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "N7ZYGGcYsRzB"
   },
   "source": [
    "We have two types of [serializers](https://spark.apache.org/docs/latest/tuning.html#data-serialization) available: \n",
    "1. Java serialization \n",
    "2. Kryo serialization\n",
    "\n",
    "Kryo is significantly faster and more compact than Java serialization (often as much as 10x), but does not support all Serializable types and requires you to register the classes you’ll use in the program in advance for best performance.\n",
    "\n",
    "Java serialization is used by default because if you have custom class that extends Serializable it can be easily used. You can also control the performance of your serialization more closely by extending java.io.Externalizable\n",
    "\n",
    "> The general recommendation is to use Kyro as the serializer whenver possible, as it leads to much smaller sizes than Java serialization. It can be added by using `.config(\"spark.serializer\", \"org.apache.spark.serializer.KryoSerializer\")` when you create your SparkSession.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "244El832wT8f"
   },
   "source": [
    "#### Shuffle Behaviour"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "XWsBFnA8xpeF"
   },
   "source": [
    "It is generally a good idea to compress the output file after the map phase. The `spark.shuffle.compress` property decides whether to do the compression or not. The compression used is `spark.io.compression.codec`.\n",
    "\n",
    "> The property can be added by using `.config(\"spark.shuffle.compress\", \"true\")` when you create your SparkSession."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "H5cFCvbczHz_"
   },
   "source": [
    "#### Compression and Serialization"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "LHU5lfyFzKw9"
   },
   "source": [
    "There are 4 defaiult codecs spark provides to compress internal data such as RDD partitions, event log, broadcast variables and shuffle outputs. They are: \n",
    "\n",
    "1. lz4\n",
    "2. lzf\n",
    "3. snappy\n",
    "4. zstd\n",
    "\n",
    "> The decision on which to use rests upon the use case. I generally use the `snappy` compression. Google created Snappy because they needed something that offered very fast compression at the expense of final size. Snappy is fast, stable and free, but it increases the size more than the other codecs. At the same time, since compute costs will be less, it seems like balanced trade off. The property can be added by using `.config(\"spark.io.compression.codec\", \"snappy\")` when you create your SparkSession.\n",
    "\n",
    "This [session](https://databricks.com/session/best-practice-of-compression-decompression-codes-in-apache-spark) explains the best practice of compression/decompression codes in Apache Spark. I recommend you to take a look at it before taking a decision."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "rdGARl-D3n-l"
   },
   "source": [
    "#### Scheduling"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Cm0ExAoS4RU6"
   },
   "source": [
    "The property `spark.speculation` performs speculative execution of tasks. This means if one or more tasks are running slowly in a stage, they will be re-launched. Speculative execution will not stop the slow running task but it launches the new task in parallel.\n",
    "\n",
    "> I usually disable this option by adding `.config(\"spark.speculation\", \"false\") ` when I create the SparkSession. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "aaxfGqYZ6Iqz"
   },
   "source": [
    "#### Application Properties"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ERovEKOU6TNE"
   },
   "source": [
    "There are mainly two application properties that you should know about:\n",
    "\n",
    "1. spark.driver.memoryOverhead - The amount of off-heap memory to be allocated per driver in cluster mode, in MiB unless otherwise specified. This is memory that accounts for things like VM overheads, interned strings, other native overheads, etc. This tends to grow with the container size (typically 6-10%). This option is currently supported on YARN and Kubernetes.\n",
    "\n",
    "2. spark.executor.memoryOverhead - The amount of off-heap memory to be allocated per executor, in MiB unless otherwise specified. This is memory that accounts for things like VM overheads, interned strings, other native overheads, etc. This tends to grow with the executor size (typically 6-10%). This option is currently supported on YARN and Kubernetes.\n",
    "\n",
    "> If you ever face an issue like `Container killed by YARN for exceeding memory limits`, know that it is because you have not specified enough memory Overhead for your job to successfully execute. The default value for Overhead is 10% of avaialbe memory (driver/executor sepearte), with minimum of 384.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "BG09dDdL6Tvt"
   },
   "source": [
    "#### Dynamic Allocation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "m_lb-JI78CVT"
   },
   "source": [
    "Lastly, I want to talk about Dynamic Allocation. This is a feature I constantly use while executing my jobs. This property is by defualt set to False. As the name suggests, it sets whether to use dynamic resource allocation, which scales the number of executors registered with this application up and down based on the workload. Truly a wonderful feature, and the greatest benefit of using it is that it will help make the best use of all the resources you have! The disadvantage of this feature is that it does not shine well when you have to execute tasks in parallel. Since most of the resources will be used by the first task, the second one will have to wait till some resource gets released. At the same time, if both get submitted at the exact same time, the resources will be shared between them, although not equally. Also, it is not guaranteed to *always* use the most optimal configurations. But in all my tests, the results have been great! \n",
    "\n",
    "> If you are planning on using this feature, you can pass the configurations as required through the spark-submit command. The four configurations which you will have to keep in mind are:<br>\n",
    "```\n",
    "--conf spark.dynamicAllocation.enabled=true\n",
    "--conf spark.dynamicAllocation.initialExecutors\n",
    "--conf spark.dynamicAllocation.minExecutors\n",
    "--conf spark.dynamicAllocation.maxExecutors\n",
    "```\n",
    "\n",
    "You can read more about this feature [here](https://spark.apache.org/docs/latest/configuration.html#dynamic-allocation) and [here](https://stackoverflow.com/questions/40200389/how-to-execute-spark-programs-with-dynamic-resource-allocation).\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "CJlmPbLYKKFA"
   },
   "source": [
    "<a id='job-tuning'></a>\n",
    "### Job Tuning"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "vccmILvVWewW"
   },
   "source": [
    "Apart from EMR and Spark tuning, there is another way to approach opttimizations, and that is by tuning your job itself to produce results efficently. I will be going over some such techniques which will help you achieve this. The [Spark Programming Guide](https://spark.apache.org/docs/2.1.1/programming-guide.html) talks more about these concepts in detail. If you guys prefer watching a video over reading, I highly recommend [A Deep Dive into Proper Optimization for Spark Jobs](https://youtu.be/daXEp4HmS-E) by Daniel Tomes from Databricks, which I found really useful and informative!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "r-R5ijHrKSg0"
   },
   "source": [
    "#### Broadcast Joins (Broadcast Hash Join)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "dvO0z5EpM5U8"
   },
   "source": [
    "For some jobs, the efficenecy can be increased by caching them in memory. Broadcast Hash Join(BHJ) is such a technique which will help you optimize join queries when the size of one side of the data is low.\n",
    ">BroadCast joins are the fastest but the drawaback is that it will consume more memory on both the executor and driver.\n",
    "\n",
    "This following steps give a sneak peek into how it works, which will help you understand the use cases where it can be used:<br>\n",
    "1. Input file(smaller of the two tables) to be broadcasted is read by the executors in parallel into its working memory.\n",
    "2. All the data from the executors is collected into driver (Hence, the need for higher memory at driver).\n",
    "3. The driver then broadcasts the combined dataset (full copy) into each executor.\n",
    "4. The size of the broadcasted dataset could be several (10-20+) times bigger the input in memory due to factors like deserialization.\n",
    "5. Executors will end up storing the parts it read first, and also the full copy, thereby leading to a high memory requirement.\n",
    "\n",
    "Some things to keep in mind about BHJ:\n",
    "1. It is advisable to use broadcast joins on small datasets only (dimesnion table, for example).\n",
    "2. Spark does not guarantee BHJ is always chosen, since not all cases (e.g. full outer join) support BHJ.\n",
    "3. You could notice skews in tasks due to uneven partition sizes; especially during aggregations, joins etc. This can be evened out by introducing Salt value (random value).<br>*Suggested formula for salt value:* random(0 – (shuffle partition count – 1))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "UOvEVcieVn7e"
   },
   "source": [
    "#### Spark Partitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ocpQiqOtVqPz"
   },
   "source": [
    "A partition in spark is an atomic chunk of data (logical division of data) stored on a node in the cluster. Partitions are the basic units of parallelism in Spark. Having too large a number of partitions or too few is not an ideal solution. The number of partitions in spark should be decided based on the cluster configuration and requirements of the application. Increasing the number of partitions will make each partition have less data or no data at all. Generally, spark partitioning can be broken down in three ways:\n",
    "1. Input\n",
    "2. Shuffle\n",
    "3. Output\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "2KY1mxZVfNsl"
   },
   "source": [
    "##### Input"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "dKw48h9eEbPa"
   },
   "source": [
    "Spark usually does a good job of figuring the ideal configuration for this one, except in very particular cases. It is advisable to use the spark default unless:\n",
    "1. Increase parallelism\n",
    "2. Heavily nested data\n",
    "3. Generating data (explode)\n",
    "4. Source is not optimal\n",
    "5. You are using UDFs\n",
    "\n",
    "`spark.sql.files.maxpartitionBytes`: This property indicates the maximum number of bytes to pack into a single partition when reading files (Default 128 MB) . Use this to increase the parallelism in reading input data. For example, if you have more cores, then you can increase the number of parallel tasks which will ensure usage of the all the cores of the cluster, and increase the speed of the task."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "qwEBu3T3EbfD"
   },
   "source": [
    "##### Shuffle"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Wx0iQUpFEbus"
   },
   "source": [
    "One of the major reason why most jobs lags in performance is, for the majority of the time, because they get the shuffle partitions count wrong. By default, the value is set to 200. In almost all situations, this is not ideal. If you are dealing with shuffle satge of less than 20 GB, 200 is fine, but otherwise this needs to be changed. For most cases, you can use the following equation to find the right value:\n",
    ">`Partition Count = Stage Input Data / Target Size` where <br>\n",
    "`Largest Shuffle Stage (Target Size) < 200MB/partition` in most cases.<br>\n",
    "`spark.sql.shuffle.partitions` property is used to set the ideal partition count value.\n",
    "\n",
    "If you ever notice that target size at the range of TBs, there is something terribly wrong, and you might want to change it back to 200, or recalculate it. Shuffle partitions can be configured for every action (not transformation) in the spark script.\n",
    "\n",
    "Let us use an example to explain this scenario: <br>\n",
    "Assume shuffle stage input = 210 GB. <br>\n",
    "Partition Count = Stage Input Data / Target Size = 210000 MB/200 MB = 1050. <br>\n",
    "As you can see, my shuffle partitions should be 1050, not 200.\n",
    "\n",
    "But, if your cluster has 2000 cores, then set your shuffle partitions to 2000.\n",
    ">In a large cluster dealing with a large data job, never set your shuffle partitions less than your total core count. \n",
    "\n",
    "\n",
    "\n",
    "Shuffle stages almost always precede the write stages and having high shuffle partition count creates small files in the output. To address this, use localCheckPoint just before write & do a coalesce call. This localCheckPoint writes the Shuffle Partition to executor local disk and then coalesces into lower partition count and hence improves the overall performance of both shuffle stage and write stage."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "-HxUYv77EdSv"
   },
   "source": [
    "##### Output"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "CPA6YRYrEdgG"
   },
   "source": [
    "There are different methods to write the data. You can control the size, composition, number of files in the output and even the number of records in each file while writing the data. While writing the data, you can increase the parallelism, thereby ensuring you use all the resources that you have. But this approach would lead to a larger number of smaller files. Usually, this isn't a problem, but if you want bigger files, you will have to use one of the compaction techniques, preferably in a cluster with lesser configuration. There are multiple ways to change the composition of the output. Keep these two in mind about composition:\n",
    "1. Coalesce: Use this to reduce the number of partitions.\n",
    "2. Repartition: Use this very rarely, and never to reduce the number of partitions<br>\n",
    "    a. Range Paritioner - It partitions the data either based on some sorted order OR set of sorted ranges of keys. <br>\n",
    "    b. Hash Partioner - It spreads around the data in the partitioning based upon the key value. Hash partitioning can make distributed data skewed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "NZ5xsdWmNOVz"
   },
   "source": [
    "<a id='best-practices'></a>\n",
    "### Best Practices"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "NUKLZ8G8NVuR"
   },
   "source": [
    "Try to incorporate these to your coding habits for better performance:\n",
    "1.   Do not use NOT IN use NOT EXISTS.\n",
    "2.   Remove Counts, Distinct Counts (use approxCountDIstinct).\n",
    "3.   Drop Duplicates early.\n",
    "4.   Always prefer SQL functions over PandasUDF.\n",
    "5.   Use Hive partitions effectively.\n",
    "6.   Leverage Spark UI effectively. \n",
    "7.   Avoid Shuffle Spills.\n",
    "8.   Aim for target cluster utilization of atleast 70%.\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "Colab and PySpark.ipynb",
   "provenance": [],
   "toc_visible": true
  },
  "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}